`zettelkasten.server`¶

zettelkasten.server ¶

Zettelkasten MCP server.

Provides tools for creating, querying, and traversing knowledge graphs stored as markdown notes with YAML frontmatter.

refresh_graphs ¶

refresh_graphs(names) -> None

Force a staleness check + reload for each named graph, bypassing the throttle.

See _refresh_graph_if_stale. Used by multi-graph, correctness-sensitive readers (cross-graph connectivity) so no in-scope graph is read stale from the in-process cache within the auto-reload throttle window. Invalid or duplicate names are skipped; only graphs whose files actually changed are reloaded.

Source code in zettelkasten/server.py

def refresh_graphs(names) -> None:
    """Force a staleness check + reload for each named graph, bypassing the throttle.

    See ``_refresh_graph_if_stale``. Used by multi-graph, correctness-sensitive
    readers (cross-graph connectivity) so no in-scope graph is read stale from the
    in-process cache within the auto-reload throttle window. Invalid or duplicate
    names are skipped; only graphs whose files actually changed are reloaded.
    """
    seen: set[str] = set()
    for name in names:
        if not name or name in seen:
            continue
        seen.add(name)
        try:
            _validate_name(name)
        except Exception:  # noqa: BLE001 -- invalid names can't be cached; skip
            continue
        _refresh_graph_if_stale(name)

create_graph ¶

create_graph(name: str, description: str, source: str = '', relations: list[dict] | None = None) -> str

Create a new concept box (knowledge graph).

Parameters:

Name	Type	Description	Default
`name`	`str`	Short kebab-case name (e.g. 'investments', 'machine-learning')	required
`description`	`str`	What this graph covers	required
`source`	`str`	Primary source material (e.g. 'Investments by Bodie, Kane, Marcus')	`''`
`relations`	`list[dict] \| None`	Optional source-level relations [{target, relation}]	`None`

Source code in zettelkasten/server.py

def create_graph(
    name: str,
    description: str,
    source: str = "",
    relations: list[dict] | None = None,
) -> str:
    """Create a new concept box (knowledge graph).

    Args:
        name: Short kebab-case name (e.g. 'investments', 'machine-learning')
        description: What this graph covers
        source: Primary source material (e.g. 'Investments by Bodie, Kane, Marcus')
        relations: Optional source-level relations [{target, relation}]
    """
    try:
        _validate_name(name)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    graph_path = GRAPHS_DIR / name
    if graph_path.exists():
        return json.dumps({"error": f"Graph '{name}' already exists.", "type": "AlreadyExists"})
    if _citation_id_taken(name):
        return json.dumps({
            "error": (f"'{name}' is already a citation id (_citations/{name}.yaml); pick a "
                      "different graph name, or promote that citation, to avoid a "
                      "source/citation name collision."),
            "type": "AlreadyExists",
        })

    if relations:
        for i, rel in enumerate(relations):
            if not isinstance(rel, dict) or "target" not in rel or "relation" not in rel:
                return json.dumps({"error": f"relations[{i}] must have 'target' and 'relation' keys.", "type": "ValidationError"})
            if rel["relation"] not in VALID_RELATIONS:
                return json.dumps({"error": f"Invalid relation '{rel['relation']}'. Must be one of: {sorted(VALID_RELATIONS)}", "type": "ValidationError"})

    graph_path.mkdir(parents=True)
    meta = graph_path / "_meta.yaml"
    import yaml
    meta_data: dict = {
        "name": name,
        "description": description,
        "source": source,
        "coverage": {"human": "none", "agent": "none"},
    }
    if relations:
        meta_data["relations"] = relations
    meta.write_text(yaml.dump(meta_data, default_flow_style=False), encoding="utf-8")

    logger.info("created graph '%s' at %s", name, graph_path)
    with _graph_lock:
        graph = ZettelGraph(name)
        # On a REVIVED name, an outgoing instance may still hold its embeddings
        # lock with an in-flight lazy .kgl build. Inherit that lock (mirroring
        # _reloaded_graph) so both builds serialize on ONE lock instead of racing
        # the same disposable cache with a fresh lock.
        prev = _graphs.get(name)
        if prev is not None:
            graph._embeddings_lock = prev._embeddings_lock
        _graphs[name] = graph
    return json.dumps({"name": name, "path": str(graph_path), "message": f"Created graph '{name}'"})

list_graphs ¶

list_graphs() -> list[dict]

List all available concept boxes (knowledge graphs).

Source graphs (no underscore prefix) are listed as primary entries. The _cross/ folder is included if it exists (synthesis notes graph).

Source code in zettelkasten/server.py

def list_graphs() -> list[dict]:
    """List all available concept boxes (knowledge graphs).

    Source graphs (no underscore prefix) are listed as primary entries.
    The _cross/ folder is included if it exists (synthesis notes graph).
    """
    results = []
    if not GRAPHS_DIR.exists():
        return results
    for d in sorted(GRAPHS_DIR.iterdir()):
        if not d.is_dir():
            continue
        # Skip _citations, _projects, _reviews (not note graphs), but include _cross
        if d.name.startswith("_") and d.name != "_cross":
            continue
        meta_path = d / "_meta.yaml"
        meta = {}
        if meta_path.exists():
            import yaml
            meta = yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
        note_count = len(list(d.glob("*.md")))
        entry = {
            "name": d.name,
            "description": meta.get("description", ""),
            "source": meta.get("source", ""),
            "note_count": note_count,
        }
        coverage = meta.get("coverage")
        if coverage:
            entry["coverage"] = coverage
        relations = meta.get("relations")
        if relations:
            entry["relations"] = relations
        results.append(entry)
    return results

add_note ¶

add_note(graph: str, title: str, type: str, body: str, source_book: str = '', source_chapter: str = '', source_page: str = '', tags: list[str] | None = None, links: list[dict] | None = None, prerequisites: list[str] | None = None, aliases: list[str] | None = None, project: str = '', synthesis_status: str = '', synthesis_graph: str = '', data: dict | None = None, snapshot: dict | None = None, grounding: dict | None = None, epistemic_status: str = '', applies_when: dict | None = None) -> dict

Add a new note to a knowledge graph.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box to add to	required
`title`	`str`	The concept/term name (e.g. 'Information Ratio')	required
`type`	`str`	One of: claim, concept, critique, definition, equation, example, figure, finding, method, model, question, quote, synthesis	required
`body`	`str`	The note content in markdown. For an `equation` note this is the canonical LaTeX (optionally `$$`-wrapped); it is checked for render-validity (balanced braces/environments, non-empty, not mojibake) before persisting and REJECTED if malformed, unless ZK_REQUIRE_EQUATION_GROUNDING is disabled. For a `figure` note this is a prose description/caption of the chart/diagram/plot.	required
`snapshot`	`dict \| None`	For an `equation` OR `figure` note, an optional visual ground-truth spec. Either a pre-built pointer (contains `path`) stored as-is, or a capture spec `{page_index?\|page?, rect?}` that renders the source PDF region to a committed sidecar PNG (see `zettelkasten.equation_snapshot`). `rect` is `[x1,y1,x2,y2]` in PDF points (a Zotero image/area annotation); omit it to snapshot the whole page. Snapshotting is best-effort — a failure warns but never blocks the note. A figure is graded `grounded` when its snapshot resolves on disk, else `inferred` (advisory, never a hard reject).	`None`
`source_book`	`str`	Book title	`''`
`source_chapter`	`str`	Chapter number or name	`''`
`source_page`	`str`	Page number	`''`
`tags`	`list[str] \| None`	Topic tags for categorization	`None`
`links`	`list[dict] \| None`	List of {target, relation, direction?} linking to other notes by ID	`None`
`prerequisites`	`list[str] \| None`	List of note IDs required to understand this	`None`
`aliases`	`list[str] \| None`	Alternative names for this concept	`None`
`data`	`dict \| None`	For a `dataset` note, the storage pointer + schema block (`{backend, format, content_hash, schema, path\|fetch}`). The numbers live out of the markdown (a committed sidecar file, a DVC-tracked file, or an API-fetched cache); see `zettelkasten.datasets`.	`None`
`project`	`str`	When adding to `_cross`, the project that should claim this synthesis note (auto-registers it into the project's `cross:` list)	`''`
`synthesis_status`	`str`	Spine lifecycle marker (`scaffold` \| `materialized`) for an apex/dimension/spec node; empty for ordinary notes.	`''`
`synthesis_graph`	`str`	With `project`, the extraction CONTEXT this write belongs to. Strict-tag enforcement resolves the source's `(project, synthesis_graph)` extraction entry (so a source extracted under several schemas/projects enforces against the right rubric); empty preserves the legacy flat-block behaviour for all existing callers. CONTRACT: when the target source carries MULTIPLE keyed extraction contexts, callers MUST thread the run's `(project, synthesis_graph)` — a context-less write degrades to the flat block (last-write-wins of the most recent structure-bearing keyed write), which may enforce against the WRONG context's rubric. The extraction trio threads this automatically (see `coordinator.extraction._structure_context`'s `CONTEXT_KEY`); a no-context resolve against a multi-context source is logged at WARNING.	`''`
`grounding`	`dict \| None`	Explicit grounding provenance for the note. Chiefly a `method='data'` citation (a re-computable statistic over a stored `dataset` note — `{method: 'data', dataset_note, derivation_id, value, ...}`) that `verify_grounding` can later re-run. Stored as-is on the note. Quote/equation grounding is computed by the server at write time and takes precedence for those note types.	`None`
`epistemic_status`	`str`	Epistemic provenance for a claim/evidence note — `grounded` (verified quote/equation citation), `measured` (a re-computable data citation), or `inferred` (asserted without a mechanical ground). Validated against `VALID_EPISTEMIC_STATUS` when provided; empty on ordinary notes.	`''`
`applies_when`	`dict \| None`	Conditional applicability for a `requirement` note (the decision-tree layer). Hybrid shape `{"condition": "<human text>", "predicate": {<param>: "<op value>"}?}` — `condition` is required when the block is present, `predicate` is optional and enables deterministic pruning by :mod:`zettelkasten.decision_tree`. None on ordinary notes.	`None`

Source code in zettelkasten/server.py

def add_note(
    graph: str,
    title: str,
    type: str,
    body: str,
    source_book: str = "",
    source_chapter: str = "",
    source_page: str = "",
    tags: list[str] | None = None,
    links: list[dict] | None = None,
    prerequisites: list[str] | None = None,
    aliases: list[str] | None = None,
    project: str = "",
    synthesis_status: str = "",
    synthesis_graph: str = "",
    data: dict | None = None,
    snapshot: dict | None = None,
    grounding: dict | None = None,
    epistemic_status: str = "",
    applies_when: dict | None = None,
) -> dict:
    """Add a new note to a knowledge graph.

    Args:
        graph: Name of the concept box to add to
        title: The concept/term name (e.g. 'Information Ratio')
        type: One of: claim, concept, critique, definition, equation, example, figure, finding, method, model, question, quote, synthesis
        body: The note content in markdown. For an ``equation`` note this is the
            canonical LaTeX (optionally ``$$``-wrapped); it is checked for
            render-validity (balanced braces/environments, non-empty, not
            mojibake) before persisting and REJECTED if malformed, unless
            ZK_REQUIRE_EQUATION_GROUNDING is disabled. For a ``figure`` note this
            is a prose description/caption of the chart/diagram/plot.
        snapshot: For an ``equation`` OR ``figure`` note, an optional visual
            ground-truth spec. Either a pre-built pointer (contains ``path``)
            stored as-is, or a capture spec ``{page_index?|page?, rect?}`` that
            renders the source PDF region to a committed sidecar PNG (see
            ``zettelkasten.equation_snapshot``). ``rect`` is ``[x1,y1,x2,y2]`` in
            PDF points (a Zotero image/area annotation); omit it to snapshot the
            whole page. Snapshotting is best-effort — a failure warns but never
            blocks the note. A figure is graded ``grounded`` when its snapshot
            resolves on disk, else ``inferred`` (advisory, never a hard reject).
        source_book: Book title
        source_chapter: Chapter number or name
        source_page: Page number
        tags: Topic tags for categorization
        links: List of {target, relation, direction?} linking to other notes by ID
        prerequisites: List of note IDs required to understand this
        aliases: Alternative names for this concept
        data: For a ``dataset`` note, the storage pointer + schema block
            (``{backend, format, content_hash, schema, path|fetch}``). The numbers
            live out of the markdown (a committed sidecar file, a DVC-tracked
            file, or an API-fetched cache); see ``zettelkasten.datasets``.
        project: When adding to ``_cross``, the project that should claim this
            synthesis note (auto-registers it into the project's ``cross:`` list)
        synthesis_status: Spine lifecycle marker (``scaffold`` | ``materialized``)
            for an apex/dimension/spec node; empty for ordinary notes.
        synthesis_graph: With ``project``, the extraction CONTEXT this write
            belongs to. Strict-tag enforcement resolves the source's
            ``(project, synthesis_graph)`` extraction entry (so a source extracted
            under several schemas/projects enforces against the right rubric);
            empty preserves the legacy flat-block behaviour for all existing
            callers. CONTRACT: when the target source carries MULTIPLE keyed
            extraction contexts, callers MUST thread the run's
            ``(project, synthesis_graph)`` — a context-less write degrades to the
            flat block (last-write-wins of the most recent structure-bearing keyed
            write), which may enforce against the WRONG context's rubric. The
            extraction trio threads this automatically (see
            ``coordinator.extraction._structure_context``'s ``CONTEXT_KEY``); a
            no-context resolve against a multi-context source is logged at WARNING.
        grounding: Explicit grounding provenance for the note. Chiefly a
            ``method='data'`` citation (a re-computable statistic over a stored
            ``dataset`` note — ``{method: 'data', dataset_note, derivation_id,
            value, ...}``) that ``verify_grounding`` can later re-run. Stored
            as-is on the note. Quote/equation grounding is computed by the server
            at write time and takes precedence for those note types.
        epistemic_status: Epistemic provenance for a claim/evidence note —
            ``grounded`` (verified quote/equation citation), ``measured`` (a
            re-computable data citation), or ``inferred`` (asserted without a
            mechanical ground). Validated against ``VALID_EPISTEMIC_STATUS`` when
            provided; empty on ordinary notes.
        applies_when: Conditional applicability for a ``requirement`` note (the
            decision-tree layer). Hybrid shape ``{"condition": "<human text>",
            "predicate": {<param>: "<op value>"}?}`` — ``condition`` is required
            when the block is present, ``predicate`` is optional and enables
            deterministic pruning by :mod:`zettelkasten.decision_tree`. None on
            ordinary notes.
    """
    zg = _get_graph(graph)
    prepared = _prepare_note(
        graph, zg, title=title, type=type, body=body,
        source_book=source_book, source_chapter=source_chapter,
        source_page=source_page, tags=tags, links=links,
        prerequisites=prerequisites, aliases=aliases, project=project,
        synthesis_status=synthesis_status, synthesis_graph=synthesis_graph,
        data=data, snapshot=snapshot, grounding=grounding,
        epistemic_status=epistemic_status, applies_when=applies_when,
    )
    if "error" in prepared:
        return prepared
    note = prepared["note"]
    filepath = zg.save_note(note)
    _maybe_advance_coverage(graph)
    # A synthesis note authored in ``_cross`` under a project scope is claimed by
    # that project (explicit membership) so it shows up in the composite.
    if graph == "_cross" and project:
        _register_project_cross(project, [note.id])
    response = prepared["response"]
    response["path"] = str(filepath)
    return response

add_notes_batch ¶

add_notes_batch(graph: str, notes: list[dict] | None = None, links: list[dict] | None = None, project: str = '', synthesis_graph: str = '') -> dict

Add many notes (and cross-links) to one graph with a SINGLE embed pass.

The batched counterpart of :func:add_note for grounded-extraction scribes, which mint many notes per slice. Each note goes through the SAME :func:_prepare_note core as :func:add_note — identical validation and the identical verify_quote grounding gate — so batching never weakens the per-note contract. Semantics are per-note best-effort: a note that fails validation/grounding fails ONLY itself (captured in failed); the rest still commit.

The efficiency win is embedding. :meth:ZettelGraph.save_note embeds each note as it is written, so a naive loop pays one model call per note. Here the box's embedding index is SUSPENDED for the duration of the writes (so each save_note performs only the crash-safe .md write + in-memory index mutation, skipping the per-note embed) and then a SINGLE batched embed pass runs over the whole batch at the end. Suspension is done under the box's _embeddings_lock so a concurrent reader never observes a half-suspended index (the same lock every index mutation already serializes on).

Parameters:

Name	Type	Description	Default
`graph`	`str`	The target box (a source graph, or `_cross`).	required
`notes`	`list[dict] \| None`	Note specs, each a dict of :func:`add_note`'s per-note kwargs (`title`/`type`/`body` + optional `tags`/`source_page`/ `aliases`/`links`/… ). `project`/`synthesis_graph` are taken from the batch-level args below (the whole batch shares one context). A spec may also carry an optional `ref` — a caller-supplied LOCAL alias (never persisted) — so a link elsewhere in this batch can point at this note before its real id is minted (see `links`).	`None`
`links`	`list[dict] \| None`	Cross-note links applied AFTER every note is saved (so both endpoints can be freshly-created notes). Each is `{from, to, relation}` (optional `direction`/`equation`/ `target_graph`); reuses :func:`link_notes` so link validation is not duplicated. Each endpoint is resolved against the batch `ref` alias map first (`ref` -> the note's minted id); a value that is not a known ref falls through unchanged, so an existing note id/title still links exactly as before. A per-note `links` list (on a note spec) is likewise deferred and ref-resolved here, with the owning note as its implicit `from` endpoint. A link naming a `ref` whose note FAILED to create is reported in `link_errors` rather than crashing.	`None`
`project`	`str`	Extraction context project (strict-tag enforcement) + the `_cross` project claim, shared by every note in the batch.	`''`
`synthesis_graph`	`str`	Extraction context synthesis graph, shared by the batch.	`''`

Returns:

Type	Description
`dict`	`{"created": [ids], "failed": [{"title", "error"}], "links_added": N}`.
`dict`	`link_errors` is added only when one or more links failed.
`dict`	`ref_warnings` is added only when a `ref` alias was ambiguous — either
`dict`	a DUPLICATE ref (two specs sharing one `ref`, last-wins) or a ref that
`dict`	COLLIDES with a note id/title that existed before the batch (a link naming
`dict`	it resolves to the batch note, shadowing the pre-existing one).

Source code in zettelkasten/server.py

def add_notes_batch(
    graph: str,
    notes: list[dict] | None = None,
    links: list[dict] | None = None,
    project: str = "",
    synthesis_graph: str = "",
) -> dict:
    """Add many notes (and cross-links) to one graph with a SINGLE embed pass.

    The batched counterpart of :func:`add_note` for grounded-extraction scribes,
    which mint many notes per slice. Each note goes through the SAME
    :func:`_prepare_note` core as :func:`add_note` — identical validation and the
    identical ``verify_quote`` grounding gate — so batching never weakens the
    per-note contract. Semantics are per-note best-effort: a note that fails
    validation/grounding fails ONLY itself (captured in ``failed``); the rest
    still commit.

    The efficiency win is embedding. :meth:`ZettelGraph.save_note` embeds each
    note as it is written, so a naive loop pays one model call per note. Here the
    box's embedding index is SUSPENDED for the duration of the writes (so each
    ``save_note`` performs only the crash-safe ``.md`` write + in-memory index
    mutation, skipping the per-note embed) and then a SINGLE batched embed pass
    runs over the whole batch at the end. Suspension is done under the box's
    ``_embeddings_lock`` so a concurrent reader never observes a half-suspended
    index (the same lock every index mutation already serializes on).

    Args:
        graph: The target box (a source graph, or ``_cross``).
        notes: Note specs, each a dict of :func:`add_note`'s per-note kwargs
            (``title``/``type``/``body`` + optional ``tags``/``source_page``/
            ``aliases``/``links``/… ). ``project``/``synthesis_graph`` are taken
            from the batch-level args below (the whole batch shares one context).
            A spec may also carry an optional ``ref`` — a caller-supplied LOCAL
            alias (never persisted) — so a link elsewhere in this batch can point
            at this note before its real id is minted (see ``links``).
        links: Cross-note links applied AFTER every note is saved (so both
            endpoints can be freshly-created notes). Each is
            ``{from, to, relation}`` (optional ``direction``/``equation``/
            ``target_graph``); reuses :func:`link_notes` so link validation is not
            duplicated. Each endpoint is resolved against the batch ``ref`` alias
            map first (``ref`` -> the note's minted id); a value that is not a
            known ref falls through unchanged, so an existing note id/title still
            links exactly as before. A per-note ``links`` list (on a note spec) is
            likewise deferred and ref-resolved here, with the owning note as its
            implicit ``from`` endpoint. A link naming a ``ref`` whose note FAILED
            to create is reported in ``link_errors`` rather than crashing.
        project: Extraction context project (strict-tag enforcement) + the
            ``_cross`` project claim, shared by every note in the batch.
        synthesis_graph: Extraction context synthesis graph, shared by the batch.

    Returns:
        ``{"created": [ids], "failed": [{"title", "error"}], "links_added": N}``.
        ``link_errors`` is added only when one or more links failed.
        ``ref_warnings`` is added only when a ``ref`` alias was ambiguous — either
        a DUPLICATE ref (two specs sharing one ``ref``, last-wins) or a ref that
        COLLIDES with a note id/title that existed before the batch (a link naming
        it resolves to the batch note, shadowing the pre-existing one).
    """
    # Fast path: a truly empty batch does no work at all — don't load the graph
    # or touch the embedding index, just return the well-formed empty result.
    if not notes and not links:
        return {"created": [], "failed": [], "links_added": 0}

    zg = _get_graph(graph)
    created: list[str] = []
    failed: list[dict] = []
    saved_notes: list = []
    cross_ids: list[str] = []
    # ``ref`` alias plumbing: map each spec's caller-supplied ``ref`` local alias
    # to the minted note id (created notes only); track refs whose note FAILED so
    # a link pointing at one is reported, not silently applied to a dangling id.
    # ``deferred_note_links`` holds each created note's own ``links`` (owner id +
    # spec) for the shared, ref-resolved post-save link phase. All three are
    # declared out here so a links-only batch (no ``notes``) still resolves.
    ref_to_id: dict[str, str] = {}
    failed_refs: set[str] = set()
    deferred_note_links: list[tuple[str, dict]] = []
    # Ref-resolution diagnostics (surfaced as ``ref_warnings`` only when non-empty
    # so the happy-path result shape is unchanged): a DUPLICATE ref (two note specs
    # sharing one ``ref``) last-wins, and a ref that COLLIDES with a note id/title
    # that already existed before this batch shadows the real note when a link
    # names it. Both silently mis-resolve otherwise; here they are reported.
    ref_warnings: list[dict] = []
    # Snapshot the pre-batch note ids so a ref colliding with a real existing id
    # can be told apart from a ref that merely equals one of THIS batch's minted
    # ids (which is the intended alias, not a collision).
    pre_batch_ids: set[str] = set(zg.notes)

    # The embed-suspension window is only entered when there are notes to write —
    # a links-only batch skips it entirely (no needless index null/restore).
    if notes:
        # Suspend the box's embedding index across the writes so each ``save_note``
        # skips its per-note embed (it no-ops the embed when ``_embeddings is
        # None``), then run one batched embed at the end. Held under
        # ``_embeddings_lock`` so a concurrent lazy build / search cannot rebuild
        # against the nulled index mid-batch — mirroring how ``delete_note`` guards
        # its index mutations.
        with zg._embeddings_lock:
            prev_index = zg._embeddings
            zg._embeddings = None
            try:
                for spec in notes:
                    if not isinstance(spec, dict):
                        failed.append({"title": "", "error": "note spec must be a mapping"})
                        continue
                    title = spec.get("title", "")
                    # A caller-supplied ``ref`` is a LOCAL alias for THIS note
                    # within the batch (not persisted) so a link can point at it
                    # before its id is minted. Its own ``links`` are DEFERRED to
                    # the post-save phase (below), not passed to ``_prepare_note``:
                    # a per-note link may reference a sibling minted later in this
                    # same batch, so embedding it now would freeze an unresolved
                    # ref alias as the raw target.
                    ref = str(spec.get("ref") or "").strip()
                    spec_links = spec.get("links")
                    # A single note's failure — validation, grounding, OR a
                    # ``save_note`` error — is isolated to that note (captured in
                    # ``failed``); the rest of the batch still commits. When the
                    # failed note carried a ``ref``, record it so a link pointing at
                    # that ref is reported rather than applied to a dangling id.
                    try:
                        prepared = _prepare_note(
                            graph, zg,
                            title=title,
                            type=spec.get("type", ""),
                            body=spec.get("body", ""),
                            source_book=spec.get("source_book", ""),
                            source_chapter=spec.get("source_chapter", ""),
                            source_page=spec.get("source_page", ""),
                            tags=spec.get("tags"),
                            links=None,
                            prerequisites=spec.get("prerequisites"),
                            aliases=spec.get("aliases"),
                            project=project,
                            synthesis_status=spec.get("synthesis_status", ""),
                            synthesis_graph=synthesis_graph,
                            data=spec.get("data"),
                            snapshot=spec.get("snapshot"),
                            grounding=spec.get("grounding"),
                            epistemic_status=spec.get("epistemic_status", ""),
                            applies_when=spec.get("applies_when"),
                        )
                        if "error" in prepared:
                            failed.append({"title": title, "error": prepared["error"]})
                            if ref:
                                failed_refs.add(ref)
                            continue
                        note = prepared["note"]
                        zg.save_note(note)
                        _maybe_advance_coverage(graph)
                    except Exception as exc:  # noqa: BLE001 — isolate one bad note
                        failed.append({"title": title, "error": str(exc)})
                        if ref:
                            failed_refs.add(ref)
                        continue
                    created.append(note.id)
                    saved_notes.append(note)
                    if ref:
                        # A repeated ref last-wins (the later note owns the alias);
                        # report it so a link intending the earlier note is not
                        # silently misrouted.
                        if ref in ref_to_id:
                            ref_warnings.append({
                                "ref": ref,
                                "warning": (
                                    f"duplicate ref '{ref}' in batch; links "
                                    "resolve to the LAST note that claimed it"
                                ),
                            })
                        ref_to_id[ref] = note.id
                    # Defer this note's own links (its id is now known) so both
                    # endpoints resolve against the completed batch ref map.
                    for lk in (spec_links or []):
                        if isinstance(lk, dict):
                            deferred_note_links.append((note.id, lk))
                    if graph == "_cross" and project:
                        cross_ids.append(note.id)
            finally:
                # Always restore the real index, even if the loop raised
                # unexpectedly, so the box is never left with a nulled index.
                zg._embeddings = prev_index

            # SINGLE batched embed pass over everything that was saved. When the
            # box was warm, embed just the new notes into the live index
            # (``index_notes`` embeds only the pending ones in one call, dropping
            # quotes); when cold, force one lazy build (loads cached vectors +
            # embeds the batch). Either way the embedder is called once for the
            # whole batch, not once per note.
            if saved_notes:
                try:
                    if zg._embeddings is not None:
                        zg._embeddings.index_notes(saved_notes)
                        zg._embeddings.save_cache_debounced()
                        # Reuse each freshly-computed vector for the global mirror
                        # (best-effort — the helper swallows its own errors).
                        for note in saved_notes:
                            zg._mirror_note_to_global(note)
                    else:
                        # Reading the property builds the index once and also warms
                        # the global mirror for the whole box.
                        _ = zg.embeddings
                except Exception as exc:  # noqa: BLE001 — writes already durable
                    # The ``.md`` files are the source of truth and are already
                    # written; only the (fully recoverable) embed decoration
                    # failed. Drop the index so the NEXT access rebuilds it cleanly
                    # from the live notes — never leave a half-updated index, and
                    # never fail a batch whose notes are safely on disk.
                    logger.warning(
                        "add_notes_batch: batched embed failed for graph '%s'; "
                        "index will rebuild on next access: %s", graph, exc,
                    )
                    zg._embeddings = None

    # A synthesis note authored in ``_cross`` under a project scope is claimed by
    # that project (explicit membership). Registered ONCE for the whole batch
    # (outside the embeddings lock) rather than per note; best-effort, never raises.
    if cross_ids:
        _register_project_cross(project, cross_ids)

    # Links are applied only after every note exists so both endpoints resolve,
    # even when both are freshly created in this batch. Each endpoint is resolved
    # against the batch ``ref`` alias map first (``ref`` -> minted id); a value
    # that is not a known ref falls through unchanged, so an existing id/title
    # still links exactly as before. An endpoint naming a ``ref`` whose note
    # FAILED to create is reported in ``link_errors`` (never applied to a dangling
    # id, never a crash). Batch-level ``links`` and each note's deferred per-note
    # ``links`` share this one resolution+application path.
    links_added = 0
    link_errors: list[dict] = []

    def _resolve_ref(value) -> "tuple[str, bool]":
        """Map a link endpoint through the batch ref alias table.

        Returns ``(resolved, failed)``: ``failed`` is True only when the value
        names a ``ref`` whose note failed to create (so the link is reported, not
        applied). A non-ref value is returned unchanged (existing id/title path).
        """
        v = str(value or "")
        if v in ref_to_id:
            return ref_to_id[v], False
        if v in failed_refs:
            return v, True
        return v, False

    def _apply_link(source_raw, target_raw, *, relation, direction, equation,
                    target_graph) -> None:
        nonlocal links_added
        rs, sfail = _resolve_ref(source_raw)
        rt, tfail = _resolve_ref(target_raw)
        if sfail or tfail:
            bad = source_raw if sfail else target_raw
            link_errors.append({
                "from": source_raw,
                "to": target_raw,
                "error": (
                    f"link references ref '{bad}', whose note failed to create in "
                    "this batch"
                ),
            })
            return
        res = link_notes(
            graph=graph,
            source_id=rs,
            target_id=rt,
            relation=relation,
            direction=direction,
            equation=equation,
            target_graph=target_graph,
        )
        parsed = json.loads(res) if isinstance(res, str) else res
        if isinstance(parsed, dict) and "error" in parsed:
            link_errors.append({"from": source_raw, "to": target_raw, "error": parsed["error"]})
        else:
            links_added += 1

    for link in links or []:
        if not isinstance(link, dict):
            link_errors.append({"link": link, "error": "link spec must be a mapping"})
            continue
        _apply_link(
            link.get("from", "") or link.get("source_id", ""),
            link.get("to", "") or link.get("target_id", ""),
            relation=link.get("relation", ""),
            direction=link.get("direction", "outgoing"),
            equation=link.get("equation", ""),
            target_graph=link.get("target_graph", ""),
        )
    # A note's own links: the owner (already minted) is the implicit ``from``; the
    # target may be a ref, an existing id, or a title. ``target``/``to`` and
    # ``graph``/``target_graph`` are both accepted (per-note vs batch link shape).
    for owner_id, lk in deferred_note_links:
        _apply_link(
            owner_id,
            lk.get("target", "") or lk.get("to", ""),
            relation=lk.get("relation", ""),
            direction=lk.get("direction", "outgoing"),
            equation=lk.get("equation", ""),
            target_graph=lk.get("graph", "") or lk.get("target_graph", ""),
        )

    # A ref that also names a note that EXISTED before this batch (by id, or by
    # title/alias) shadows that real note: a link naming the ref resolves to the
    # freshly-minted batch note, not the pre-existing one the caller may have
    # meant. Report each such collision (skipping refs that resolve to one of this
    # batch's own notes, which is the intended alias, not a collision).
    created_set = set(created)
    for ref, minted_id in ref_to_id.items():
        collides = False
        if ref in pre_batch_ids:
            collides = True
        else:
            hit = zg.find_by_title(ref)
            if hit is not None and getattr(hit, "id", None) not in created_set:
                collides = True
        if collides:
            ref_warnings.append({
                "ref": ref,
                "warning": (
                    f"ref '{ref}' collides with an existing note id/title; a link "
                    f"naming it resolves to the batch note ({minted_id}), NOT the "
                    "pre-existing note"
                ),
            })

    result = {"created": created, "failed": failed, "links_added": links_added}
    if link_errors:
        result["link_errors"] = link_errors
    if ref_warnings:
        result["ref_warnings"] = ref_warnings
    return result

attach_to_dimension ¶

attach_to_dimension(graph: str, note_id: str, tag: str, project: str = '', synthesis_graph: str = '') -> dict

Attach a claim to its schema dimension node in the run's spine — deterministically.

The whole point: the scribe names only the dimension tag; the SERVER resolves the dimension node id, the synthesis graph, the relation, and — most importantly — the EDGE DIRECTION from the run's structure persisted on the source meta (extraction.structure). The scribe never decides direction, which is the spine-wiring failure mode a smaller model is most likely to get wrong.

Direction is keyed on the persisted attach_relation: - spine-member (fresh scaffold default + promoted spine): SPINE-SIDE — dim_node --spine-member--> claim, edge lives in the SYNTHESIS graph, cross-graph into the source. This is the ONLY direction the v2 matrix/outline/overlay membership readback sees. - input-to (explicit legacy base-side): BASE-SIDE — claim --input-to--> dim_node, edge lives in the SOURCE graph, cross-graph into the synthesis graph. HARDENING: because the membership readback indexes ONLY spine-side spine-member edges, a base-side attach is additionally MIRRORED into a canonical dim_node --spine-member--> claim edge so the claim is never membership-invisible. The return carries mirrored_membership in this case.

Returns the edge it wrote, or an error dict (no structure for this source, an unknown tag, or a failed link). The caller (the note dispatch) holds the write lock over BOTH graphs.

project + synthesis_graph select the extraction CONTEXT whose structure is used to resolve the dimension node. A source attached to several spines (multiple schemas/projects) keeps an independent structure per context. Resolution follows :func:resolve_extraction_context:

With NEITHER supplied (no-context), the flat block's structure is used (a keyed write mirrors its coherent rubric -- including structure -- into the flat block, so a no-context attach against a keyed-written source still resolves a spine).
With BOTH supplied (a FULL key), the matching (project, synthesis_graph) context's structure is used; a populated contexts list with no match surfaces a clean ExtractionContextNotFound (never a silent mis-route to another context's structure).
A PARTIAL pair (only one supplied) can never match a keyed entry, so it degrades to the flat block exactly like a no-context read -- it does NOT raise. (The trio never passes a partial pair: scribe/auditor pass both keys or neither.)

CONTRACT: against a source carrying MULTIPLE keyed contexts, a no-context attach resolves the flat block's structure (the LAST structure-bearing keyed write, last-write-wins) and may route the claim onto the WRONG spine. Non-trio / manual callers MUST thread the run's (project, synthesis_graph); such a degrade is logged at WARNING by :func:resolve_extraction_context.

Source code in zettelkasten/server.py

def attach_to_dimension(
    graph: str,
    note_id: str,
    tag: str,
    project: str = "",
    synthesis_graph: str = "",
) -> dict:
    """Attach a claim to its schema dimension node in the run's spine — deterministically.

    The whole point: the scribe names only the dimension ``tag``; the SERVER
    resolves the dimension node id, the synthesis graph, the relation, and — most
    importantly — the EDGE DIRECTION from the run's structure persisted on the
    source meta (``extraction.structure``). The scribe never decides direction,
    which is the spine-wiring failure mode a smaller model is most likely to get
    wrong.

    Direction is keyed on the persisted ``attach_relation``:
      - ``spine-member`` (fresh scaffold default + promoted spine): SPINE-SIDE —
        dim_node --spine-member--> claim, edge lives in the SYNTHESIS graph,
        cross-graph into the source. This is the ONLY direction the v2
        matrix/outline/overlay membership readback sees.
      - ``input-to`` (explicit legacy base-side): BASE-SIDE — claim --input-to-->
        dim_node, edge lives in the SOURCE graph, cross-graph into the synthesis
        graph. HARDENING: because the membership readback indexes ONLY spine-side
        ``spine-member`` edges, a base-side attach is additionally MIRRORED into a
        canonical ``dim_node --spine-member--> claim`` edge so the claim is never
        membership-invisible. The return carries ``mirrored_membership`` in this
        case.

    Returns the edge it wrote, or an error dict (no structure for this source, an
    unknown ``tag``, or a failed link). The caller (the ``note`` dispatch) holds
    the write lock over BOTH graphs.

    ``project`` + ``synthesis_graph`` select the extraction CONTEXT whose
    ``structure`` is used to resolve the dimension node. A source attached to
    several spines (multiple schemas/projects) keeps an independent structure per
    context. Resolution follows :func:`resolve_extraction_context`:

    * With NEITHER supplied (no-context), the flat block's ``structure`` is used
      (a keyed write mirrors its coherent rubric -- including structure -- into the
      flat block, so a no-context attach against a keyed-written source still
      resolves a spine).
    * With BOTH supplied (a FULL key), the matching ``(project, synthesis_graph)``
      context's ``structure`` is used; a populated ``contexts`` list with no match
      surfaces a clean ``ExtractionContextNotFound`` (never a silent mis-route to
      another context's structure).
    * A PARTIAL pair (only one supplied) can never match a keyed entry, so it
      degrades to the flat block exactly like a no-context read -- it does NOT
      raise. (The trio never passes a partial pair: scribe/auditor pass both keys
      or neither.)

    CONTRACT: against a source carrying MULTIPLE keyed contexts, a no-context
    attach resolves the flat block's ``structure`` (the LAST structure-bearing
    keyed write, last-write-wins) and may route the claim onto the WRONG spine.
    Non-trio / manual callers MUST thread the run's ``(project, synthesis_graph)``;
    such a degrade is logged at WARNING by :func:`resolve_extraction_context`.
    """
    try:
        ext = resolve_extraction_context(
            graph, project=project or None, synthesis_graph=synthesis_graph or None,
        ) or {}
    except ExtractionContextNotFound as exc:
        # Explicit context demanded but this source has no entry for it: do not
        # silently resolve a DIFFERENT context's structure (which would attach the
        # claim onto the wrong spine). Surface a clean tool error.
        return {
            "error": str(exc) + (
                " Pass the (project, synthesis_graph) of THIS run's spine (one of "
                "the available contexts), or omit both to use the source's flat "
                "structure."
            ),
            "type": "ExtractionContextNotFound",
            "project": exc.project,
            "synthesis_graph": exc.synthesis_graph,
            "available_contexts": exc.available,
        }
    structure = ext.get("structure") if isinstance(ext, dict) else None
    if not isinstance(structure, dict) or not structure.get("dimension_nodes"):
        return {
            "error": (
                f"Source '{graph}' has no materialized spine structure to attach "
                "to. This is a flat run — link the claim to the HUB instead."
            ),
            "type": "BadRequest",
        }
    dimension_nodes = structure.get("dimension_nodes") or {}
    node_id = dimension_nodes.get(tag)
    if not node_id:
        return {
            "error": (
                f"Dimension tag '{tag}' has no node in this run's spine. Use one "
                "of the materialized dimension tags."
            ),
            "type": "NotFound",
            "dimension_tags": sorted(dimension_nodes),
        }
    synthesis_graph = str(structure.get("synthesis_graph") or "")
    attach_relation = str(structure.get("attach_relation") or "input-to")
    if not synthesis_graph:
        return {"error": "Run structure is missing its synthesis graph.", "type": "BadRequest"}

    mirrored_membership = False
    if attach_relation == "spine-member":
        # SPINE-SIDE: the edge originates on the dimension node in the synthesis
        # graph and points cross-graph at the claim in the source graph.
        res = link_notes(
            graph=synthesis_graph, source_id=node_id, target_id=note_id,
            relation="spine-member", target_graph=graph,
        )
        edge = f"{node_id} --spine-member--> {note_id}"
        edge_graph = synthesis_graph
    else:
        # BASE-SIDE: the edge originates on the claim in the source graph and
        # points cross-graph at the dimension node in the synthesis graph.
        res = link_notes(
            graph=graph, source_id=note_id, target_id=node_id,
            relation=attach_relation, target_graph=synthesis_graph,
        )
        edge = f"{note_id} --{attach_relation}--> {node_id}"
        edge_graph = graph

    parsed = json.loads(res) if isinstance(res, str) else res
    if isinstance(parsed, dict) and "error" in parsed:
        return parsed

    if attach_relation != "spine-member":
        # HARDENING — a spine is NEVER membership-invisible ("never
        # input-to-only"). The v2 membership readback (``_spine_membership_index``
        # / the spine overlay) indexes ONLY spine-side ``spine-member`` edges, so a
        # base-side attach relation (``input-to``) that renders in the base-side
        # matrix column fallback is INVISIBLE in the overlay — the exact divergence
        # that stranded grounded-extraction spines. Mirror every base-side attach
        # into the canonical spine-side membership edge (``dim --spine-member-->
        # claim@source``) so both readbacks agree by construction. Additive (the
        # base-side semantic edge above is left intact) and idempotent (``link_notes``
        # dedups; an AlreadyExists is "present"). A failed mirror is logged, not
        # fatal — the base-side edge already committed.
        mres = link_notes(
            graph=synthesis_graph, source_id=node_id, target_id=note_id,
            relation="spine-member", target_graph=graph,
        )
        mparsed = json.loads(mres) if isinstance(mres, str) else mres
        mirrored_membership = not (
            isinstance(mparsed, dict)
            and mparsed.get("error")
            and mparsed.get("type") != "AlreadyExists"
        )
        if not mirrored_membership:
            logger.warning(
                "attach_to_dimension: base-side %s attach %s->%s committed but the "
                "spine-member mirror failed (%s); this claim will be invisible in "
                "the spine overlay/membership readback.",
                attach_relation, note_id, node_id, mparsed,
            )

    out = {
        "attached": True,
        "tag": tag,
        "dimension_node": node_id,
        "relation": attach_relation,
        "edge": edge,
        "edge_graph": edge_graph,
        "synthesis_graph": synthesis_graph,
    }
    if attach_relation != "spine-member":
        # Surface that the canonical spine-side membership edge was also written
        # (the base-side relation alone would be overlay-invisible).
        out["mirrored_membership"] = mirrored_membership
    return out

link_notes ¶

link_notes(graph: str, source_id: str, target_id: str, relation: str, direction: str = 'outgoing', equation: str = '', target_graph: str = '', primary: bool = False) -> str

Add a link between two existing notes.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`source_id`	`str`	ID of the note to add the link FROM	required
`target_id`	`str`	ID of the note being linked TO	required
`relation`	`str`	Relationship type (e.g. 'depends-on', 'contrasts', 'supports')	required
`direction`	`str`	'outgoing' or 'bidirectional'	`'outgoing'`
`equation`	`str`	Optional formula capturing the mathematical relationship (e.g. 'IR = α / ω')	`''`
`target_graph`	`str`	Optional target graph name for cross-graph links (e.g. another source name, '_citations', '_cross')	`''`
`primary`	`bool`	Mark the edge as THE one durable structural parent of a cross-graph sub-spine composition (`component-of` apex→apex). Left False on every ordinary edge.	`False`

Source code in zettelkasten/server.py

def link_notes(
    graph: str,
    source_id: str,
    target_id: str,
    relation: str,
    direction: str = "outgoing",
    equation: str = "",
    target_graph: str = "",
    primary: bool = False,
) -> str:
    """Add a link between two existing notes.

    Args:
        graph: Name of the concept box
        source_id: ID of the note to add the link FROM
        target_id: ID of the note being linked TO
        relation: Relationship type (e.g. 'depends-on', 'contrasts', 'supports')
        direction: 'outgoing' or 'bidirectional'
        equation: Optional formula capturing the mathematical relationship (e.g. 'IR = α / ω')
        target_graph: Optional target graph name for cross-graph links (e.g. another source name, '_citations', '_cross')
        primary: Mark the edge as THE one durable structural parent of a
            cross-graph sub-spine composition (``component-of`` apex→apex). Left
            False on every ordinary edge.
    """
    if relation not in VALID_RELATIONS:
        return json.dumps({"error": f"Invalid relation '{relation}'. Must be one of: {sorted(VALID_RELATIONS)}", "type": "ValidationError"})

    if target_graph:
        # Valid link targets are NOTE graphs (real source folders + ``_cross``)
        # plus ``_citations`` (citation links are legitimate). Overlay folders
        # (``_reviews``, ``_projects``, and any other ``_``-prefixed folder that
        # isn't ``_cross``/``_citations``) hold YAML manifests, not notes, so a
        # link into them would be dangling and unresolvable — reject them even
        # though ``is_dir()`` is true for them.
        valid_special = {"_citations", "_cross"}
        is_overlay = target_graph.startswith("_") and target_graph not in valid_special
        if is_overlay or (
            target_graph not in valid_special and not (GRAPHS_DIR / target_graph).is_dir()
        ):
            return json.dumps({"error": f"Target graph '{target_graph}' does not exist.", "type": "NotFound"})

    zg = _get_graph(graph)
    source_note = zg.notes.get(source_id)
    if not source_note:
        return json.dumps({"error": f"Source note '{source_id}' not found in graph '{graph}'.", "type": "NotFound"})

    if not target_graph and target_id not in zg.notes:
        return json.dumps({"error": f"Target note '{target_id}' not found in graph '{graph}'.", "type": "NotFound"})

    for existing in source_note.links:
        if existing.target == target_id and existing.relation == relation and existing.graph == target_graph:
            # A build-time sub-spine seed onto a PRE-EXISTING (unflagged)
            # ``component-of`` edge must be able to PROMOTE it to ``primary`` —
            # otherwise the seed reports success (connected=True) while the edge
            # stays primary=False and the child is silently un-nested. Refresh the
            # flag in place (matching ``spine.SpineGraphOps.ensure_link``) and
            # report success. Only ever UPGRADES to primary when requested; an
            # ordinary re-link (primary defaulting False) never demotes an existing
            # primary edge.
            if primary and not bool(getattr(existing, "primary", False)):
                existing.primary = True
                zg.save_note(source_note)
                return json.dumps({"source": source_id, "target": target_id, "relation": relation, "target_graph": target_graph, "message": f"Refreshed primary flag: {source_note.title} --[{relation}]--> {target_id}"})
            return json.dumps({"error": f"Link already exists: {source_id} --[{relation}]--> {target_id}", "type": "AlreadyExists"})

    source_note.links.append(Link(target=target_id, relation=relation, direction=direction, equation=equation, graph=target_graph, primary=bool(primary)))
    zg.save_note(source_note)

    target_title = "(cross-graph)" if target_graph else zg.notes[target_id].title
    return json.dumps({"source": source_id, "target": target_id, "relation": relation, "target_graph": target_graph, "message": f"Linked: {source_note.title} --[{relation}]--> {target_title}"})

get_note ¶

get_note(graph: str, note_id: str = '', title: str = '') -> dict

Get a note by ID or by title/alias, including full content and all links.

Also useful for checking whether a concept already has a note before creating a new one (pass title).

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	The note's unique ID (preferred when known)	`''`
`title`	`str`	Title or alias to look up (case-insensitive; used when note_id is empty)	`''`

Source code in zettelkasten/server.py

def get_note(graph: str, note_id: str = "", title: str = "") -> dict:
    """Get a note by ID or by title/alias, including full content and all links.

    Also useful for checking whether a concept already has a note before
    creating a new one (pass title).

    Args:
        graph: Name of the concept box
        note_id: The note's unique ID (preferred when known)
        title: Title or alias to look up (case-insensitive; used when note_id
            is empty)
    """
    zg = _get_graph(graph)
    if not note_id and not title:
        return {"error": "Provide note_id or title.", "type": "BadRequest"}
    note = zg.notes.get(note_id) if note_id else zg.find_by_title(title)
    if not note:
        wanted = note_id or title
        return {"found": False, "error": f"Note '{wanted}' not found in graph '{graph}'.", "type": "NotFound"}

    # Explicit read: record an access so retrieval can boost recently-used notes.
    _record_access(graph, note.id, weight=1.0)

    return {
        "id": note.id,
        "title": note.title,
        "type": note.type,
        "source": note.source,
        "body": note.body,
        "tags": note.tags,
        "links": [
            {k: v for k, v in [
                ("target", l.target),
                ("relation", l.relation),
                ("direction", l.direction),
                ("equation", l.equation),
                ("graph", l.graph),
            ] if v}
            for l in note.links
        ],
        "prerequisites": note.prerequisites,
        "aliases": note.aliases,
        "backlinks": zg.get_backlinks(note.id),
    }

search_notes ¶

search_notes(graph: str, query: str = '', type: str = '', tag: str = '', mode: str = 'hybrid', top_k: int = 20, limit: int = 50) -> list[dict]

Search notes by meaning + keyword, or list all notes when query is empty.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`query`	`str`	Search term. Empty = list all notes (with filters).	`''`
`type`	`str`	Optional filter by note type	`''`
`tag`	`str`	Optional filter by tag	`''`
`mode`	`str`	"hybrid" (default) fuses keyword + semantic rankings via Reciprocal Rank Fusion — the best general default: exact term matches rank at the top while conceptually related notes are still surfaced. Use "keyword" for exact-substring/id lookups or bulk type/tag browsing (returns up to `limit`, no embedding cost), or "semantic" for pure embedding search.	`'hybrid'`
`top_k`	`int`	Semantic/hybrid mode only: number of results (default 20)	`20`
`limit`	`int`	Keyword/list mode: cap on results returned (default 50) so large graphs don't return thousands of notes at once. Pass 0 for no cap; narrow with type/tag/query or raise the limit if you need more.	`50`

Source code in zettelkasten/server.py

def search_notes(
    graph: str,
    query: str = "",
    type: str = "",
    tag: str = "",
    mode: str = "hybrid",
    top_k: int = 20,
    limit: int = 50,
) -> list[dict]:
    """Search notes by meaning + keyword, or list all notes when query is empty.

    Args:
        graph: Name of the concept box
        query: Search term. Empty = list all notes (with filters).
        type: Optional filter by note type
        tag: Optional filter by tag
        mode: "hybrid" (default) fuses keyword + semantic rankings via Reciprocal
            Rank Fusion — the best general default: exact term matches rank at the
            top while conceptually related notes are still surfaced. Use
            "keyword" for exact-substring/id lookups or bulk type/tag browsing
            (returns up to `limit`, no embedding cost), or "semantic" for pure
            embedding search.
        top_k: Semantic/hybrid mode only: number of results (default 20)
        limit: Keyword/list mode: cap on results returned (default 50) so large
            graphs don't return thousands of notes at once. Pass 0 for no cap;
            narrow with type/tag/query or raise the limit if you need more.
    """
    if mode == "semantic" and query.strip():
        from zettelkasten import config as zk_config

        cfg = zk_config.rerank_config()
        disabled = zk_config.rerank_disabled(cfg)
        recall_k = top_k if disabled else max(top_k, cfg.overfetch * top_k)
        results_sem = semantic_search(graph, query, top_k=recall_k)
        if disabled:
            results_sem = results_sem[:top_k]
        else:
            results_sem = _rerank_note_dicts(graph, results_sem, top_k=top_k, cfg=cfg)
        if type:
            results_sem = [r for r in results_sem if r.get("type") == type]
        return results_sem

    if mode == "hybrid" and query.strip():
        return _hybrid_search_notes(graph, query, type=type, top_k=top_k)

    zg = _get_graph(graph)

    # When no post-filters are applied, push the cap down into the search/scan so
    # we stop early; otherwise cap after filtering.
    cap = limit if limit and limit > 0 else None
    inner_limit = cap if not (type or tag) else None

    if query:
        results = zg.search(query, limit=inner_limit)
    else:
        results = list(zg.notes.values())

    if type:
        results = [n for n in results if n.type == type]
    if tag:
        results = [n for n in results if tag in n.tags]

    if cap is not None:
        results = results[:cap]

    return [
        {
            "id": n.id,
            "title": n.title,
            "type": n.type,
            "tags": n.tags,
            "body_preview": n.body[:150] + "..." if len(n.body) > 150 else n.body,
        }
        for n in results
    ]

list_notes ¶

list_notes(graph: str, type: str = '', tag: str = '') -> list[dict]

List all notes in a graph with optional filters.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`type`	`str`	Filter by note type (definition, concept, model, etc.)	`''`
`tag`	`str`	Filter by tag	`''`

Source code in zettelkasten/server.py

def list_notes(graph: str, type: str = "", tag: str = "") -> list[dict]:
    """List all notes in a graph with optional filters.

    Args:
        graph: Name of the concept box
        type: Filter by note type (definition, concept, model, etc.)
        tag: Filter by tag
    """
    zg = _get_graph(graph)
    return zg.list_notes(type_filter=type or None, tag=tag or None)

follow_links ¶

follow_links(graph: str, note_id: str, relation: str = '', direction: str = 'both', project: str = '') -> dict

Traverse a note's links, filtered by relation and/or direction.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	The note to traverse from	required
`relation`	`str`	Optional relation filter (e.g. 'depends-on', 'input-to')	`''`
`direction`	`str`	'outgoing' \| 'incoming' \| 'both' (default). 'incoming' returns only backlinks; 'outgoing' only forward links.	`'both'`
`project`	`str`	When set with an incoming direction, ALSO gather cross-graph backlinks from every source graph in the project — the one call that rolls up a synthesis dimension node's grounding (the claims attaching to it from other graphs).	`''`

Source code in zettelkasten/server.py

def follow_links(
    graph: str,
    note_id: str,
    relation: str = "",
    direction: str = "both",
    project: str = "",
) -> dict:
    """Traverse a note's links, filtered by relation and/or direction.

    Args:
        graph: Name of the concept box
        note_id: The note to traverse from
        relation: Optional relation filter (e.g. 'depends-on', 'input-to')
        direction: 'outgoing' | 'incoming' | 'both' (default). 'incoming' returns
            only backlinks; 'outgoing' only forward links.
        project: When set with an incoming direction, ALSO gather cross-graph
            backlinks from every source graph in the project — the one call that
            rolls up a synthesis dimension node's grounding (the claims attaching
            to it from other graphs).
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found.", "type": "NotFound"}

    # Explicit read: following a note's links is an intentional use of it.
    _record_access(graph, note.id, weight=1.0)

    rel = relation or None
    want_out = direction in ("outgoing", "both", "")
    want_in = direction in ("incoming", "both", "")

    result: dict = {"note": note.title}
    if want_out:
        result["outgoing_links"] = zg.get_linked(note_id, relation=rel)
    if want_in:
        backlinks = zg.get_backlinks(note_id)
        if rel is not None:
            backlinks = [b for b in backlinks if b.get("relation") == rel]
        result["backlinks"] = backlinks
        if project:
            result["cross_graph_backlinks"] = _project_backlinks(
                project, graph, note_id, rel
            )
    return result

get_prerequisites ¶

get_prerequisites(graph: str, note_id: str) -> dict

Get the full prerequisite chain for a note (what you need to know first).

Returns prerequisites in learning order (earliest first).

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	The note to get prerequisites for	required

Source code in zettelkasten/server.py

def get_prerequisites(graph: str, note_id: str) -> dict:
    """Get the full prerequisite chain for a note (what you need to know first).

    Returns prerequisites in learning order (earliest first).

    Args:
        graph: Name of the concept box
        note_id: The note to get prerequisites for
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found.", "type": "NotFound"}

    chain = zg.get_prerequisites_chain(note_id)
    ordered = []
    for nid in chain:
        n = zg.notes.get(nid)
        if n:
            ordered.append({"id": n.id, "title": n.title, "type": n.type})

    return {
        "note": note.title,
        "prerequisite_chain": ordered,
    }

find_by_title ¶

find_by_title(graph: str, title: str) -> dict

Look up a note by its title or alias (case-insensitive).

Useful for checking if a concept already has a note before creating a new one.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`title`	`str`	The title or alias to search for	required

Source code in zettelkasten/server.py

def find_by_title(graph: str, title: str) -> dict:
    """Look up a note by its title or alias (case-insensitive).

    Useful for checking if a concept already has a note before creating a new one.

    Args:
        graph: Name of the concept box
        title: The title or alias to search for
    """
    zg = _get_graph(graph)
    note = zg.find_by_title(title)
    if not note:
        return {"found": False, "message": f"No note found for '{title}'."}

    return {
        "found": True,
        "id": note.id,
        "title": note.title,
        "type": note.type,
        "tags": note.tags,
    }

get_graph_summary ¶

get_graph_summary(graph: str) -> dict

Get a summary of a knowledge graph: counts by type, most-linked notes, orphans.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required

Source code in zettelkasten/server.py

def get_graph_summary(graph: str) -> dict:
    """Get a summary of a knowledge graph: counts by type, most-linked notes, orphans.

    Args:
        graph: Name of the concept box
    """
    zg = _get_graph(graph)

    type_counts: dict[str, int] = {}
    link_counts: dict[str, int] = {}
    linked_to: dict[str, int] = {}

    for note in zg.notes.values():
        type_counts[note.type] = type_counts.get(note.type, 0) + 1
        link_counts[note.id] = len(note.links)
        for link in note.links:
            linked_to[link.target] = linked_to.get(link.target, 0) + 1

    # Find orphans (no incoming or outgoing links)
    orphans = [
        {"id": n.id, "title": n.title}
        for n in zg.notes.values()
        if not n.links and n.id not in linked_to
    ]

    # Most connected notes
    all_connections = {
        nid: link_counts.get(nid, 0) + linked_to.get(nid, 0)
        for nid in zg.notes
    }
    most_connected = sorted(all_connections.items(), key=lambda x: x[1], reverse=True)[:10]

    return {
        "name": graph,
        "total_notes": len(zg.notes),
        "by_type": type_counts,
        "most_connected": [
            {"id": nid, "title": zg.notes[nid].title, "connections": count}
            for nid, count in most_connected
        ],
        "orphans": orphans[:20],
    }

health ¶

health() -> str

Return diagnostic information about the zettelkasten server.

Reports loaded graphs, note counts, embedding index status, graphs directory path, and install location.

Source code in zettelkasten/server.py

@mcp_server.tool()
def health() -> str:
    """Return diagnostic information about the zettelkasten server.

    Reports loaded graphs, note counts, embedding index status,
    graphs directory path, and install location.
    """
    graphs_info = []
    if GRAPHS_DIR.exists():
        for d in sorted(GRAPHS_DIR.iterdir()):
            if not d.is_dir():
                continue
            if d.name.startswith("_") and d.name != "_cross":
                continue
            note_count = len(list(d.glob("*.md")))
            cached = d.name in _graphs
            has_embeddings = False
            if cached and _graphs[d.name]._embeddings is not None:
                has_embeddings = True
            graphs_info.append({
                "name": d.name,
                "note_count": note_count,
                "cached": cached,
                "embeddings_loaded": has_embeddings,
            })

    total_notes = sum(g["note_count"] for g in graphs_info)
    global_index: dict = {"available": False}
    try:
        from zettelkasten import config as zk_config
        from zettelkasten.global_index import global_index_status

        global_index = global_index_status()
        global_index["available"] = True
        global_index["use_global_index"] = zk_config.use_global_index()
        indexed = global_index.get("indexed_notes")
        if indexed is not None and total_notes:
            # Coverage vs the local note total (excludes quote-type notes, which are
            # not embedded, so <100% is expected on a quote-heavy corpus).
            global_index["coverage_ratio"] = round(min(1.0, indexed / total_notes), 3)
    except Exception as exc:  # noqa: BLE001 - health must never fail on the ANN
        global_index = {"available": False, "error": str(exc)}

    return json.dumps({
        "graphs_dir": str(GRAPHS_DIR),
        "graphs_dir_exists": GRAPHS_DIR.exists(),
        "graphs": graphs_info,
        "total_graphs": len(graphs_info),
        "total_notes": total_notes,
        "total_cached": sum(1 for g in graphs_info if g["cached"]),
        "global_index": global_index,
        "install_location": str(Path(__file__).resolve().parent),
        "auto_sync_interval_seconds": _AUTO_SYNC_INTERVAL,
    })

rebuild_global_index ¶

rebuild_global_index() -> str

Rebuild the global note-level ANN index from the live .md sources.

The global index (_global.kgl) is a disposable derived cache that powers broad, cross-project semantic search (project_search / query_topic) with a single pre-filtered vector query instead of a per-box fan-out. It is maintained incrementally as notes are written, but this tool does a full rebuild — reconciling any drift (e.g. notes deleted while the server was down) and re-deriving every vector from the source of truth. Unchanged notes reuse their existing vector, so a rebuild does not re-embed the whole corpus.

Source code in zettelkasten/server.py

@mcp_server.tool()
def rebuild_global_index() -> str:
    """Rebuild the global note-level ANN index from the live ``.md`` sources.

    The global index (``_global.kgl``) is a disposable derived cache that powers
    broad, cross-project semantic search (``project_search`` / ``query_topic``)
    with a single pre-filtered vector query instead of a per-box fan-out. It is
    maintained incrementally as notes are written, but this tool does a full
    rebuild — reconciling any drift (e.g. notes deleted while the server was down)
    and re-deriving every vector from the source of truth. Unchanged notes reuse
    their existing vector, so a rebuild does not re-embed the whole corpus.
    """
    try:
        from zettelkasten.global_index import get_global_index

        gi = get_global_index()
        if gi is None:
            return json.dumps({
                "error": "Global index unavailable (semantic search disabled).",
                "type": "Unavailable",
            })
        summary = gi.rebuild(graphs_dir=GRAPHS_DIR)
        return json.dumps({"message": "Rebuilt global ANN index.", **summary})
    except Exception as exc:  # noqa: BLE001
        return json.dumps({"error": str(exc), "type": "RebuildError"})

semantic_search ¶

semantic_search(graph: str, query: str, top_k: int = 10) -> list[dict]

Search notes by meaning using embeddings (not just keyword matching).

Use this when keyword search fails or you want conceptually related notes even if they don't share exact terms.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`query`	`str`	Natural language query (e.g. 'measures of risk-adjusted performance')	required
`top_k`	`int`	Number of results to return (default 10)	`10`

Source code in zettelkasten/server.py

def semantic_search(graph: str, query: str, top_k: int = 10) -> list[dict]:
    """Search notes by meaning using embeddings (not just keyword matching).

    Use this when keyword search fails or you want conceptually related notes
    even if they don't share exact terms.

    Args:
        graph: Name of the concept box
        query: Natural language query (e.g. 'measures of risk-adjusted performance')
        top_k: Number of results to return (default 10)
    """
    zg = _get_graph(graph)
    if not zg.notes:
        return []

    results = zg.embeddings.search(query, top_k=top_k)
    return [
        {
            "id": note_id,
            "title": zg.notes[note_id].title,
            "type": zg.notes[note_id].type,
            "similarity": round(score, 3),
            "body_preview": zg.notes[note_id].body[:150] + "..."
            if len(zg.notes[note_id].body) > 150
            else zg.notes[note_id].body,
        }
        for note_id, score in results
        if note_id in zg.notes
    ]

suggest_connections ¶

suggest_connections(graph: str, note_id: str, top_k: int = 8) -> dict

Find notes that are semantically related but not yet linked.

Use this after creating a note to discover potential connections, or periodically to find missing links in the graph.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	The note to find connections for	required
`top_k`	`int`	Number of suggestions to return	`8`

Source code in zettelkasten/server.py

def suggest_connections(graph: str, note_id: str, top_k: int = 8) -> dict:
    """Find notes that are semantically related but not yet linked.

    Use this after creating a note to discover potential connections,
    or periodically to find missing links in the graph.

    Args:
        graph: Name of the concept box
        note_id: The note to find connections for
        top_k: Number of suggestions to return
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found.", "type": "NotFound"}

    # Get IDs already linked
    already_linked = {link.target for link in note.links}
    already_linked.update(note.prerequisites)
    # Also exclude notes that link back to this one
    for other in zg.notes.values():
        for link in other.links:
            if link.target == note_id:
                already_linked.add(other.id)

    similar = zg.embeddings.find_similar(note_id, top_k=top_k + len(already_linked), exclude_ids=already_linked)

    suggestions = []
    for other_id, score in similar[:top_k]:
        other = zg.notes.get(other_id)
        if not other:
            continue
        suggestions.append({
            "id": other.id,
            "title": other.title,
            "type": other.type,
            "similarity": round(score, 3),
            "body_preview": other.body[:100],
        })

    return {
        "note": note.title,
        "suggestions": suggestions,
        "message": f"Found {len(suggestions)} potentially related notes not yet linked to '{note.title}'.",
    }

suggest_structure ¶

suggest_structure(graph: str) -> dict

Analyze the graph and suggest structural improvements.

Proactively identifies: - Orphan notes (no links in or out) that should probably connect to something - Dense clusters that might benefit from a synthesis note - Frequently referenced concepts that don't have their own note yet - Notes with high semantic similarity that aren't linked

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required

Source code in zettelkasten/server.py

def suggest_structure(graph: str) -> dict:
    """Analyze the graph and suggest structural improvements.

    Proactively identifies:
    - Orphan notes (no links in or out) that should probably connect to something
    - Dense clusters that might benefit from a synthesis note
    - Frequently referenced concepts that don't have their own note yet
    - Notes with high semantic similarity that aren't linked

    Args:
        graph: Name of the concept box
    """
    zg = _get_graph(graph)
    if len(zg.notes) < 3:
        return {"message": "Not enough notes yet for structural analysis. Keep adding!"}

    # Find orphans
    linked_to: set[str] = set()
    has_outgoing: set[str] = set()
    for note in zg.notes.values():
        if note.links:
            has_outgoing.add(note.id)
        for link in note.links:
            linked_to.add(link.target)

    orphans = [
        {"id": n.id, "title": n.title, "type": n.type}
        for n in zg.notes.values()
        if n.id not in linked_to and n.id not in has_outgoing
    ]

    # Find potential missing links (high similarity, no link)
    missing_links = []
    seen_pairs: set[tuple[str, str]] = set()
    for note in list(zg.notes.values())[:50]:  # cap for performance
        similar = zg.embeddings.find_similar(note.id, top_k=3)
        existing_targets = {l.target for l in note.links}
        for other_id, score in similar:
            if score < 0.6:
                continue
            if other_id in existing_targets:
                continue
            pair = tuple(sorted([note.id, other_id]))
            if pair in seen_pairs:
                continue
            seen_pairs.add(pair)
            other = zg.notes.get(other_id)
            if other:
                missing_links.append({
                    "note_a": {"id": note.id, "title": note.title},
                    "note_b": {"id": other.id, "title": other.title},
                    "similarity": round(score, 3),
                })

    # Sort missing links by similarity
    missing_links.sort(key=lambda x: x["similarity"], reverse=True)

    # Identify clusters by tag co-occurrence
    large_clusters = _structure_clusters(zg)

    # Find referenced-but-missing concepts (terms in link targets that don't resolve)
    unresolved = set()
    for note in zg.notes.values():
        for link in note.links:
            if link.target not in zg.notes:
                unresolved.add(link.target)

    suggestions = []
    if orphans:
        suggestions.append(f"Found {len(orphans)} orphan note(s) with no connections — consider linking them.")
    if missing_links:
        suggestions.append(f"Found {len(missing_links)} pairs of semantically similar notes that aren't linked.")
    if large_clusters:
        suggestions.append(f"Found {len(large_clusters)} topic cluster(s) large enough for synthesis notes.")
    if unresolved:
        suggestions.append(f"Found {len(unresolved)} referenced note(s) that don't exist yet — consider creating them.")

    return {
        "summary": suggestions or ["Graph looks well-connected!"],
        "orphans": orphans[:10],
        "missing_links": missing_links[:10],
        "clusters_for_synthesis": large_clusters[:5],
        "unresolved_references": sorted(unresolved)[:10],
    }

overview ¶

overview() -> str

Use this first when you're unsure how to work with the zettelkasten: a compact orientation map.

Read-only. Returns the store's purpose, the workflows you reach for most, and a tool -> action index. Authoritative action values also live on each tool's own action schema enum. Reads are always available; writes may be gated by ZK_DISABLE_WRITE / ZK_DISABLE_DELETE.

Source code in zettelkasten/server.py

@mcp_server.tool()
def overview() -> str:
    """Use this first when you're unsure how to work with the zettelkasten: a compact orientation map.

    Read-only. Returns the store's purpose, the workflows you reach for most,
    and a tool -> action index. Authoritative action values also live on each
    tool's own ``action`` schema enum. Reads are always available; writes may be
    gated by ``ZK_DISABLE_WRITE`` / ``ZK_DISABLE_DELETE``.
    """
    return (
        "# Zettelkasten server — orientation\n\n"
        "A grounded knowledge graph: sources (papers/docs) -> notes + claims ->\n"
        "spines/graphs, with citations and cross-project synthesis. Reads are\n"
        "always available; writes may be gated (ZK_DISABLE_WRITE / ZK_DISABLE_DELETE).\n\n"
        "## Common workflows\n"
        "- Explore a corpus: `syllabus`, `note(action=\"search\"/\"get\")`, `synapse(action=\"search\")`.\n"
        "- Ingest / extract: `source(action=\"ingest\"/\"create\")`, `schema`, `remine`.\n"
        "- Synthesize: `synapse(action=\"frame\"/\"connections\"/\"synthesize\")`, `claim(action=\"propose\"/\"create\")`.\n"
        "- Research a question: `query_topic`, `frame_question`, `guide`.\n\n"
        "## Tools (action-dispatch)\n"
        "- `note(get|search|follow|prerequisites|add|add_batch|update|link|attach|verify_grounding)`\n"
        "- `source(fulltext|outline|verify|coverage|protocol|ingest|create|set_coverage|backfill|mark_extracted)`\n"
        "- `dataset(add|get|values|snapshot|attach)`\n"
        "- `graph(list|summary|project|create|export|import)`\n"
        "- `citation(create|dedup|promote|expand_corpus|expand|unlink|export_bibliography)`\n"
        "- `project(create|add|add_cross|remove_cross|set_default_spine)` [write]\n"
        "- `spine(list|promote|demote|resync|verify|delete)`\n"
        "- `schema(list|get|spines|validate|save)`\n"
        "- `remine(propose|apply)`\n"
        "- `zotero(lookup|collections|list|fetch_pdf)`\n"
        "- `review(list|load|create|update|rename|delete|suggest_claims|propose_placements)`\n"
        "- `claim(create|delete|delete_theme|materialize|materialize_theme|propose|test|discover_contradictions|debate_map|find_supersessions)`\n"
        "- `syllabus(papers|claims|outline|gaps|missing|anatomy)`\n"
        "- `synapse(search|frame|get|connections|matrix|synthesize|debate|build)`\n"
        "- `guide(apply|next)`\n\n"
        "## Single-purpose tools\n"
        "`suggest`, `query_topic`, `frame_question`, `citation`, `health`,\n"
        "`rebuild_global_index`, `launch_dashboard`.\n\n"
        "## Resources you can pull\n"
        "- `zettelkasten://overview` (this map) · `zettelkasten://vocabulary` "
        "(valid note types/relations/statuses — read before guessing a value).\n"
    )

suggest ¶

suggest(kind: str, graph: str = '', note_id: str = '', top_k: int = 8, project: str = '', similarity_threshold: 'float | None' = None, min_sources: int = 2, since_source: str = '', accept: dict | None = None, target_scope: str = 'project', include_types: 'list[str] | None' = None, min_cross_degree: int = 1, max_candidates: int = 500, max_notes_per_source: int = 200) -> dict

Use this when you want the graph analyzed proactively: suggest links, structure fixes, or synthesis hubs.

Kinds

connections: Notes semantically related to note_id but not yet linked, WITHIN note_id's own graph. Use after creating a note. Requires graph and note_id. (For CROSS-graph candidates use cross-connections.) cross-connections: Semantically related notes in a DIFFERENT graph that are not yet linked, plus a connectivity audit (island graphs + under-connected notes). This is the job-2 (flat cross-graph connectivity) engine — the cross-graph counterpart of connections, which is graph-local and cannot connect a note to another source. READ-ONLY: write the edges you judge real with note(action="link", ..., target_graph=). Scope with note_id (+graph), graph, or project (or none = whole corpus); target_scope (default 'project') restricts candidate targets to a project's sources or opens them to the whole corpus. Idempotent: already-linked pairs are excluded, so a re-run after wiring edges returns only what's missing. structure: Whole-graph analysis — orphans, missing links, clusters ready for synthesis notes, unresolved references. Requires graph. concept-hubs: Cross-source concept clusters in a project that could become synthesis notes in _cross/. Requires project. Each suggestion is tagged action='attach' (link into an existing hub — preferred) or action='new_hub' (mint a new synthesis note). Pass since_source after adding/extracting a source to get the cheap incremental check that only looks at that source (and follows its links into existing hubs). Pass accept (a proposed theme: {title, members, theme_id?, body?, tags?}) to ACCEPT a proposed theme — it materializes a landscape concept hub in _cross/ (a concept note tagged landscape + surveys edges) via the idempotent, ownership-safe materialize_theme producer wrapper and returns the hub.

Parameters:

Name	Type	Description	Default
`kind`	`str`	"connections", "cross-connections", "structure", or "concept-hubs".	required
`graph`	`str`	Concept box name (connections, structure, cross-connections).	`''`
`note_id`	`str`	Note to find connections for (connections, cross-connections).	`''`
`top_k`	`int`	Number of suggestions / candidate targets per source note.	`8`
`project`	`str`	Project name (concept-hubs, cross-connections).	`''`
`similarity_threshold`	`'float \| None'`	Minimum pair similarity. Omit to use the kind's own default (concept-hubs 0.7; cross-connections 0.5 — cross-graph cosine is compressed, so genuine neighbors sit ~0.5–0.66). Pass a value to take control: raise it for precision, lower it for recall.	`None`
`min_sources`	`int`	Minimum distinct sources per cluster (concept-hubs).	`2`
`since_source`	`str`	Restrict concept-hubs to clusters involving this one source (incremental, cheap — use right after adding or extracting it).	`''`
`accept`	`dict \| None`	Accept a proposed theme (concept-hubs): a dict with `title` (the concept label), `members` (work refs the hub `surveys`), and an optional `theme_id` / `body` / `tags`. Materializes the landscape hub instead of returning suggestions.	`None`
`target_scope`	`str`	cross-connections — 'project' (default; targets restricted to the project's sources) or 'corpus' (any source graph).	`'project'`
`include_types`	`'list[str] \| None'`	cross-connections — content note types to consider (default claim/finding/method/definition/model/concept; never quotes or hub/spine nodes).	`None`
`min_cross_degree`	`int`	cross-connections — a content note with fewer cross-graph edges than this is reported `under_connected`.	`1`
`max_candidates`	`int`	cross-connections — global cap on returned pairs (`truncated` flags an overflow).	`500`
`max_notes_per_source`	`int`	cross-connections — per-graph note cap for the similarity sweep (bounds cost/memory as the corpus grows).	`200`

Source code in zettelkasten/server.py

@mcp_server.tool()
def suggest(
    kind: str,
    graph: str = "",
    note_id: str = "",
    top_k: int = 8,
    project: str = "",
    similarity_threshold: "float | None" = None,
    min_sources: int = 2,
    since_source: str = "",
    accept: dict | None = None,
    target_scope: str = "project",
    include_types: "list[str] | None" = None,
    min_cross_degree: int = 1,
    max_candidates: int = 500,
    max_notes_per_source: int = 200,
) -> dict:
    """Use this when you want the graph analyzed proactively: suggest links, structure fixes, or synthesis hubs.

    Kinds:
        connections: Notes semantically related to note_id but not yet linked,
            WITHIN note_id's own graph. Use after creating a note. Requires graph
            and note_id. (For CROSS-graph candidates use ``cross-connections``.)
        cross-connections: Semantically related notes in a DIFFERENT graph that
            are not yet linked, plus a connectivity audit (island graphs +
            under-connected notes). This is the job-2 (flat cross-graph
            connectivity) engine — the cross-graph counterpart of ``connections``,
            which is graph-local and cannot connect a note to another source.
            READ-ONLY: write the edges you judge real with
            note(action="link", ..., target_graph=<b.graph>). Scope with note_id
            (+graph), graph, or project (or none = whole corpus); ``target_scope``
            (default 'project') restricts candidate targets to a project's sources
            or opens them to the whole corpus. Idempotent: already-linked pairs are
            excluded, so a re-run after wiring edges returns only what's missing.
        structure: Whole-graph analysis — orphans, missing links, clusters
            ready for synthesis notes, unresolved references. Requires graph.
        concept-hubs: Cross-source concept clusters in a project that could
            become synthesis notes in _cross/. Requires project. Each suggestion
            is tagged action='attach' (link into an existing hub — preferred) or
            action='new_hub' (mint a new synthesis note). Pass since_source after
            adding/extracting a source to get the cheap incremental check that
            only looks at that source (and follows its links into existing hubs).
            Pass ``accept`` (a proposed theme: ``{title, members, theme_id?,
            body?, tags?}``) to ACCEPT a proposed theme — it materializes a
            landscape concept hub in _cross/ (a ``concept`` note tagged
            ``landscape`` + ``surveys`` edges) via the idempotent, ownership-safe
            ``materialize_theme`` producer wrapper and returns the hub.

    Args:
        kind: "connections", "cross-connections", "structure", or "concept-hubs".
        graph: Concept box name (connections, structure, cross-connections).
        note_id: Note to find connections for (connections, cross-connections).
        top_k: Number of suggestions / candidate targets per source note.
        project: Project name (concept-hubs, cross-connections).
        similarity_threshold: Minimum pair similarity. Omit to use the kind's own
            default (concept-hubs 0.7; cross-connections 0.5 — cross-graph cosine
            is compressed, so genuine neighbors sit ~0.5–0.66). Pass a value to
            take control: raise it for precision, lower it for recall.
        min_sources: Minimum distinct sources per cluster (concept-hubs).
        since_source: Restrict concept-hubs to clusters involving this one source
            (incremental, cheap — use right after adding or extracting it).
        accept: Accept a proposed theme (concept-hubs): a dict with ``title`` (the
            concept label), ``members`` (work refs the hub ``surveys``), and an
            optional ``theme_id`` / ``body`` / ``tags``. Materializes the
            landscape hub instead of returning suggestions.
        target_scope: cross-connections — 'project' (default; targets restricted
            to the project's sources) or 'corpus' (any source graph).
        include_types: cross-connections — content note types to consider
            (default claim/finding/method/definition/model/concept; never quotes
            or hub/spine nodes).
        min_cross_degree: cross-connections — a content note with fewer cross-graph
            edges than this is reported ``under_connected``.
        max_candidates: cross-connections — global cap on returned pairs
            (``truncated`` flags an overflow).
        max_notes_per_source: cross-connections — per-graph note cap for the
            similarity sweep (bounds cost/memory as the corpus grows).
    """
    if kind == "connections":
        if not graph or not note_id:
            return {"error": "kind='connections' requires graph and note_id.", "type": "BadRequest"}
        return suggest_connections(graph, note_id, top_k=top_k)
    if kind == "cross-connections":
        # Cross-graph cosine range is compressed (genuine neighbors sit ~0.5–0.66),
        # so this kind defaults LOWER than concept-hubs. The agent stays in full
        # control: pass similarity_threshold to raise it for precision or drop it
        # for recall. None → the kind's own default.
        thr = 0.5 if similarity_threshold is None else similarity_threshold
        return json.loads(suggest_cross_connections(
            project=project, graph=graph, note_id=note_id, top_k=top_k,
            similarity_threshold=thr, target_scope=target_scope,
            include_types=include_types, min_cross_degree=min_cross_degree,
            max_candidates=max_candidates, max_notes_per_source=max_notes_per_source,
        ))
    if kind == "structure":
        if not graph:
            return {"error": "kind='structure' requires graph.", "type": "BadRequest"}
        return suggest_structure(graph)
    if kind == "concept-hubs":
        # Acceptance path: materialize a landscape hub from a proposed theme. This
        # is the only WRITE in suggest — it reuses the idempotent, ownership-safe
        # materialize_theme producer wrapper (which refuses federated targets and
        # never clobbers a non-concept note of the same id).
        if accept is not None:
            # RUNTIME write barrier: ``suggest`` is intentionally always
            # registered (its suggestion path is read-only), but the ``accept``
            # branch performs a GRAPH WRITE via ``materialize_theme``. Guard it
            # here so a prompt-injected agent cannot mutate the graph through the
            # always-registered tool when ZK_DISABLE_WRITE is set.
            if _WRITE_DISABLED:
                return {"error": "writes are disabled for this server instance", "type": "WriteDisabled"}
            from zettelkasten import claim_producer

            title = str(accept.get("title") or "").strip()
            if not title:
                return {"error": "accept requires a 'title' (the concept label).", "type": "BadRequest"}
            try:
                hub = claim_producer.materialize_theme(
                    _get_graph,
                    theme_id=str(accept.get("theme_id") or "").strip(),
                    title=title,
                    members=accept.get("members"),
                    body=accept.get("body"),
                    tags=accept.get("tags"),
                )
            except Exception as e:
                return json.loads(_json_tool_error(e))
            # Strong usage: accepting a theme materializes a landscape hub — a
            # deliberate synthesis action — so boost the hub's access-recency.
            if isinstance(hub, dict) and hub.get("id"):
                _record_access(
                    str(hub.get("graph") or ""), str(hub.get("id")), weight=3.0
                )
            return hub
        if not project:
            project, err = _resolve_project(project)
            if err:
                return err
        hub_thr = 0.7 if similarity_threshold is None else similarity_threshold
        return json.loads(suggest_concept_hubs(
            project, similarity_threshold=hub_thr,
            min_sources=min_sources, since_source=since_source,
        ))
    return {"error": f"Unknown kind '{kind}'. Use 'connections', 'cross-connections', 'structure', or 'concept-hubs'.", "type": "BadRequest"}

update_note ¶

update_note(graph: str, note_id: str, append_body: str = '', new_type: str = '', add_tags: list[str] | None = None, add_links: list[dict] | None = None, add_aliases: list[str] | None = None, add_prerequisites: list[str] | None = None, replace_body: str = '', synthesis_status: str = '') -> dict

Update an existing note with additional content or metadata.

Use this when the user provides more depth about an existing concept (additional formulas, interpretations, caveats, etc.)

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	ID of the note to update	required
`append_body`	`str`	Markdown content to append to the body	`''`
`replace_body`	`str`	Overwrite the body wholesale (mutually exclusive with append_body). Use this when a synthesizer authors a spine node's portrait so the original "claims attach here" scaffold sentence is replaced, not left stranded above the prose.	`''`
`new_type`	`str`	Upgrade the note type (e.g. 'definition' → 'concept')	`''`
`add_tags`	`list[str] \| None`	Tags to add (won't duplicate existing ones)	`None`
`add_links`	`list[dict] \| None`	New links to add: [{target, relation, direction?}]	`None`
`add_aliases`	`list[str] \| None`	New aliases to add	`None`
`add_prerequisites`	`list[str] \| None`	New prerequisite IDs to add	`None`
`synthesis_status`	`str`	Flip a spine node's lifecycle (`scaffold` \| `materialized`); set `materialized` when its prose is authored.	`''`

Source code in zettelkasten/server.py

def update_note(
    graph: str,
    note_id: str,
    append_body: str = "",
    new_type: str = "",
    add_tags: list[str] | None = None,
    add_links: list[dict] | None = None,
    add_aliases: list[str] | None = None,
    add_prerequisites: list[str] | None = None,
    replace_body: str = "",
    synthesis_status: str = "",
) -> dict:
    """Update an existing note with additional content or metadata.

    Use this when the user provides more depth about an existing concept
    (additional formulas, interpretations, caveats, etc.)

    Args:
        graph: Name of the concept box
        note_id: ID of the note to update
        append_body: Markdown content to append to the body
        replace_body: Overwrite the body wholesale (mutually exclusive with
            append_body). Use this when a synthesizer authors a spine node's
            portrait so the original "claims attach here" scaffold sentence is
            replaced, not left stranded above the prose.
        new_type: Upgrade the note type (e.g. 'definition' → 'concept')
        add_tags: Tags to add (won't duplicate existing ones)
        add_links: New links to add: [{target, relation, direction?}]
        add_aliases: New aliases to add
        add_prerequisites: New prerequisite IDs to add
        synthesis_status: Flip a spine node's lifecycle (``scaffold`` |
            ``materialized``); set ``materialized`` when its prose is authored.
    """
    if append_body and replace_body:
        return {"error": "Pass either append_body or replace_body, not both.", "type": "BadRequest"}
    if synthesis_status and synthesis_status not in VALID_SYNTHESIS_STATUS:
        return {
            "error": (
                f"Invalid synthesis_status '{synthesis_status}'. "
                f"Must be one of: {sorted(VALID_SYNTHESIS_STATUS)}"
            ),
            "type": "BadRequest",
        }

    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found in graph '{graph}'.", "type": "NotFound"}

    changed = []

    if replace_body:
        note.body = replace_body.strip()
        changed.append("body (replaced)")
    elif append_body:
        note.body = note.body.rstrip() + "\n\n" + append_body
        changed.append("body")

    if new_type:
        if new_type not in VALID_TYPES:
            return {"error": f"Invalid type '{new_type}'.", "type": "BadRequest"}
        if new_type != note.type:
            changed.append(f"type: {note.type} → {new_type}")
            note.type = new_type

    if add_tags:
        for tag in add_tags:
            if tag not in note.tags:
                note.tags.append(tag)
                changed.append(f"+tag:{tag}")

    if add_links:
        for link_data in add_links:
            rel = link_data.get("relation", "")
            if rel and rel not in VALID_RELATIONS:
                return {"error": f"Invalid relation '{rel}'.", "type": "BadRequest"}
            link_graph = link_data.get("graph", "")
            existing = {(l.target, l.relation, l.graph) for l in note.links}
            if (link_data["target"], rel, link_graph) not in existing:
                note.links.append(Link(
                    target=link_data["target"],
                    relation=rel,
                    direction=link_data.get("direction", "outgoing"),
                    equation=link_data.get("equation", ""),
                    graph=link_graph,
                ))
                changed.append(f"+link:{rel}→{link_data['target']}")

    if add_aliases:
        for alias in add_aliases:
            if alias not in note.aliases:
                note.aliases.append(alias)
                changed.append(f"+alias:{alias}")

    if add_prerequisites:
        for prereq in add_prerequisites:
            if prereq not in note.prerequisites:
                note.prerequisites.append(prereq)
                changed.append(f"+prereq:{prereq}")

    if synthesis_status and synthesis_status != note.synthesis_status:
        changed.append(f"synthesis_status: {note.synthesis_status or '(none)'} → {synthesis_status}")
        note.synthesis_status = synthesis_status

    if not changed:
        return {"message": "No changes made.", "id": note_id}

    filepath = zg.save_note(note)
    return {
        "id": note_id,
        "title": note.title,
        "changes": changed,
        "path": str(filepath),
    }

verify_grounding ¶

verify_grounding(graph: str, note_id: str) -> dict

Re-verify a note's method='data' grounding by RE-COMPUTING it.

Loads the note and re-runs :func:zettelkasten.data_grounding.verify_data_grounding over its data citation. On a CLEAN reproduce (verified True, severity ok) it stamps grounding.verified=True plus verified_at (UTC ISO 8601) and verified_in (<repo>@<short-sha> for this repo) onto the note and saves it. The structured verify result is ALWAYS returned, whether or not the claim reproduced. angelo VERIFIES/STORES here — it never composes the claim.

Source code in zettelkasten/server.py

def verify_grounding(graph: str, note_id: str) -> dict:
    """Re-verify a note's ``method='data'`` grounding by RE-COMPUTING it.

    Loads the note and re-runs
    :func:`zettelkasten.data_grounding.verify_data_grounding` over its data
    citation. On a CLEAN reproduce (``verified`` True, ``severity`` ``ok``) it
    stamps ``grounding.verified=True`` plus ``verified_at`` (UTC ISO 8601) and
    ``verified_in`` (``<repo>@<short-sha>`` for this repo) onto the note and
    saves it. The structured verify result is ALWAYS returned, whether or not the
    claim reproduced. angelo VERIFIES/STORES here — it never composes the claim.
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found in graph '{graph}'.", "type": "NotFound"}

    grounding = getattr(note, "grounding", None)
    if not isinstance(grounding, dict) or grounding.get("method") != "data":
        return {
            "error": (
                "Note has no method='data' grounding to verify. verify_grounding "
                "re-computes a data citation; quote/equation grounding is verified "
                "at write time."
            ),
            "type": "GroundingError",
        }

    from zettelkasten.data_grounding import verify_data_grounding

    result = verify_data_grounding(note, zg.path)

    if result.get("verified") and result.get("severity") == "ok":
        from datetime import datetime, timezone

        verified_at = datetime.now(timezone.utc).isoformat()
        verified_in = _repo_stamp(zg.path)
        grounding["verified"] = True
        grounding["verified_at"] = verified_at
        grounding["verified_in"] = verified_in
        note.grounding = grounding
        filepath = zg.save_note(note)
        result["verified_at"] = verified_at
        result["verified_in"] = verified_in
        result["saved"] = True
        result["path"] = str(filepath)

    return result

set_note_confidence ¶

set_note_confidence(graph: str, note_id: str, score: float, *, method: str = 'logprob') -> dict

Attach an advisory extraction-confidence score to a note's grounding.

Confidence is ADVISORY (governance): it is merged onto note.grounding (preserving any quote-verification keys already there) and surfaced in the note panel + synthesis-matrix cells, but it never gates publishing or routing. score is clamped to 0..1. This is the persistence seam for the Stream post-extraction scoring pass (:mod:stream.scoring); it is a plain in-process mutator (no MCP surface) — callers serialize writes themselves.

Parameters:

Name	Type	Description	Default
`graph`	`str`	Source graph the note lives in.	required
`note_id`	`str`	Note to annotate (typically a `claim`/`finding`).	required
`score`	`float`	Support/extraction confidence in 0..1 (a >1 value is treated as a percent and divided by 100).	required
`method`	`str`	How the score was produced (e.g. `logprob` \| `sampling`).	`'logprob'`

Source code in zettelkasten/server.py

def set_note_confidence(
    graph: str,
    note_id: str,
    score: float,
    *,
    method: str = "logprob",
) -> dict:
    """Attach an advisory extraction-confidence score to a note's grounding.

    Confidence is ADVISORY (governance): it is merged onto ``note.grounding``
    (preserving any quote-verification keys already there) and surfaced in the
    note panel + synthesis-matrix cells, but it never gates publishing or
    routing. ``score`` is clamped to 0..1. This is the persistence seam for the
    Stream post-extraction scoring pass (:mod:`stream.scoring`); it is a plain
    in-process mutator (no MCP surface) — callers serialize writes themselves.

    Args:
        graph: Source graph the note lives in.
        note_id: Note to annotate (typically a ``claim``/``finding``).
        score: Support/extraction confidence in 0..1 (a >1 value is treated as a
            percent and divided by 100).
        method: How the score was produced (e.g. ``logprob`` | ``sampling``).
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return {"error": f"Note '{note_id}' not found in graph '{graph}'.", "type": "NotFound"}
    try:
        s = float(score)
    except (TypeError, ValueError):
        return {"error": f"score must be a number, got {score!r}.", "type": "BadRequest"}
    if s > 1.0:
        s = s / 100.0
    s = max(0.0, min(1.0, s))
    grounding = dict(note.grounding) if isinstance(note.grounding, dict) else {}
    grounding["score"] = s
    if method:
        grounding["method"] = method
    note.grounding = grounding
    filepath = zg.save_note(note)
    return {"id": note_id, "grounding": grounding, "path": str(filepath)}

delete_note ¶

delete_note(graph: str, note_id: str) -> str

Delete a note from the knowledge graph.

Removes the markdown file and de-indexes the note. Does NOT remove links from other notes that point to this one (they become unresolved references, visible via suggest_structure).

Parameters:

Name	Type	Description	Default
`graph`	`str`	Name of the concept box	required
`note_id`	`str`	ID of the note to delete	required

Source code in zettelkasten/server.py

@_delete_tool()
def delete_note(graph: str, note_id: str) -> str:
    """Delete a note from the knowledge graph.

    Removes the markdown file and de-indexes the note. Does NOT remove
    links from other notes that point to this one (they become unresolved
    references, visible via suggest_structure).

    Args:
        graph: Name of the concept box
        note_id: ID of the note to delete
    """
    zg = _get_graph(graph)
    note = zg.notes.get(note_id)
    if not note:
        return json.dumps({"error": f"Note '{note_id}' not found in graph '{graph}'.", "type": "NotFound"})

    title = note.title
    filepath = zg.path / f"{note_id}.md"

    if filepath.exists():
        filepath.unlink()

    # Serialize the index mutations against any concurrent lazy build / save_note
    # on the same shared graph instance (kglite is not thread-safe and another
    # threadpool call may be snapshotting zg.notes right now).
    with zg._embeddings_lock:
        del zg.notes[note_id]
        zg._title_index.pop(title.lower(), None)
        for alias in note.aliases:
            zg._alias_index.pop(alias.lower(), None)
        for tag in note.tags:
            if tag in zg._tag_index:
                zg._tag_index[tag].discard(note_id)

        if zg._embeddings is not None:
            zg._embeddings.remove_note(note_id)
            zg._embeddings.save_cache()

    # Mirror the deletion into the global ANN index (best-effort; a global-index
    # failure must never break the .md delete). Skips federated graphs.
    if not getattr(zg, "repo_id", None):
        try:
            from zettelkasten.global_index import get_global_index

            gi = get_global_index()
            if gi is not None:
                gi.remove_note(graph, note_id)
        except Exception as exc:  # pragma: no cover - best-effort
            logger.debug("Global index: could not remove note '%s': %s", note_id, exc)

    # Same write-time bundle pre-warm signal as save_note: a deletion changes the
    # box's source signature, so a long-lived dashboard that opted in can rebuild
    # the affected project bundle(s) proactively rather than waiting for a poll to
    # notice the stale signature. Best-effort / no-op when pre-warm is disabled.
    from zettelkasten.graph import _signal_note_written

    _signal_note_written(graph)

    logger.info("deleted note '%s' (%s) from graph '%s'", title, note_id, graph)
    return json.dumps({"id": note_id, "title": title, "message": f"Deleted '{title}' ({note_id})."})

create_source ¶

create_source(name: str, doc_type: str, title: str = '', authors: list[str] | None = None, year: int | None = None, venue: str = '', doi: str = '', abstract: str = '', zotero_key: str = '', content_hash: str = '', source_path: str = '', relations: list[dict] | None = None, update: bool = False, date: str = '') -> str

Create a source folder with enriched _meta.yaml.

Dedups on content_hash: if a source already carries the same content_hash (the sha256 from ingest_source), the existing source is returned instead of creating a duplicate.

GROWABLE SOURCES: pass update=True for a source whose underlying document grows over time (e.g. a live transcript streamed segment by segment). Then an EXISTING box of this name is not an error — its content pointer (content_hash / source_path) is REFRESHED to the new (larger) document and the memoized grounding corpus is invalidated, so subsequent incremental extraction reads and grounds against the whole accumulated text while all claims land in the one box. With update=True the content-hash dedup shortcut is skipped (the caller is deliberately targeting this named box).

Parameters:

Name	Type	Description	Default
`name`	`str`	Short kebab-case folder name (e.g. 'bodie-investments-12e')	required
`doc_type`	`str`	Document type (e.g. 'journal-article', 'book', 'working-paper', 'report')	required
`title`	`str`	Full title of the source	`''`
`authors`	`list[str] \| None`	List of author names	`None`
`year`	`int \| None`	Publication year	`None`
`venue`	`str`	Journal, publisher, or conference	`''`
`doi`	`str`	Digital Object Identifier	`''`
`abstract`	`str`	Brief abstract or description	`''`
`zotero_key`	`str`	Zotero item key for reference manager integration	`''`
`content_hash`	`str`	sha256 of the source file's bytes (from ingest_source); anchors the full-text reference and dedups re-ingested files	`''`
`source_path`	`str`	Path the full text was extracted from (local path or Zotero PDF path)	`''`
`relations`	`list[dict] \| None`	Optional source-level relations [{target, relation}]	`None`
`date`	`str`	Optional fine-grained source date (`YYYY-MM-DD`, `YYYY-MM`, or a full ISO datetime) — finer than `year`. Used to stamp the global ANN index and drive calendar-recency reranking (e.g. an earnings call's date). Falls back to `year` when absent.	`''`

Source code in zettelkasten/server.py

def create_source(
    name: str,
    doc_type: str,
    title: str = "",
    authors: list[str] | None = None,
    year: int | None = None,
    venue: str = "",
    doi: str = "",
    abstract: str = "",
    zotero_key: str = "",
    content_hash: str = "",
    source_path: str = "",
    relations: list[dict] | None = None,
    update: bool = False,
    date: str = "",
) -> str:
    """Create a source folder with enriched _meta.yaml.

    Dedups on content_hash: if a source already carries the same content_hash
    (the sha256 from ingest_source), the existing source is returned instead of
    creating a duplicate.

    GROWABLE SOURCES: pass ``update=True`` for a source whose underlying document
    grows over time (e.g. a live transcript streamed segment by segment). Then an
    EXISTING box of this ``name`` is not an error — its content pointer
    (``content_hash`` / ``source_path``) is REFRESHED to the new (larger) document
    and the memoized grounding corpus is invalidated, so subsequent incremental
    extraction reads and grounds against the whole accumulated text while all
    claims land in the one box. With ``update=True`` the content-hash dedup shortcut
    is skipped (the caller is deliberately targeting this named box).

    Args:
        name: Short kebab-case folder name (e.g. 'bodie-investments-12e')
        doc_type: Document type (e.g. 'journal-article', 'book', 'working-paper', 'report')
        title: Full title of the source
        authors: List of author names
        year: Publication year
        venue: Journal, publisher, or conference
        doi: Digital Object Identifier
        abstract: Brief abstract or description
        zotero_key: Zotero item key for reference manager integration
        content_hash: sha256 of the source file's bytes (from ingest_source); anchors the full-text reference and dedups re-ingested files
        source_path: Path the full text was extracted from (local path or Zotero PDF path)
        relations: Optional source-level relations [{target, relation}]
        date: Optional fine-grained source date (``YYYY-MM-DD``, ``YYYY-MM``, or a
            full ISO datetime) — finer than ``year``. Used to stamp the global ANN
            index and drive calendar-recency reranking (e.g. an earnings call's
            date). Falls back to ``year`` when absent.
    """
    try:
        _validate_name(name)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    if content_hash and not update:
        existing = _find_source_by_hash(content_hash)
        if existing:
            return json.dumps({
                "name": existing,
                "deduped": True,
                "matched_on": content_hash,
                "message": f"A source for this content already exists as '{existing}'; reusing it.",
            })

    graph_path = GRAPHS_DIR / name
    if graph_path.exists():
        if update:
            return _refresh_source_content(
                name, graph_path, content_hash=content_hash,
                source_path=source_path, title=title, doc_type=doc_type,
                date=date,
            )
        return json.dumps({"error": f"Source '{name}' already exists.", "type": "AlreadyExists"})
    if _citation_id_taken(name):
        return json.dumps({
            "error": (f"'{name}' is already a citation id (_citations/{name}.yaml); promote that "
                      "citation (citation action='promote') instead of creating a same-named "
                      "source, to avoid a source/citation name collision."),
            "type": "AlreadyExists",
        })

    if relations:
        for i, rel in enumerate(relations):
            if not isinstance(rel, dict) or "target" not in rel or "relation" not in rel:
                return json.dumps({"error": f"relations[{i}] must have 'target' and 'relation' keys.", "type": "ValidationError"})
            if rel["relation"] not in VALID_RELATIONS:
                return json.dumps({"error": f"Invalid relation '{rel['relation']}'. Must be one of: {sorted(VALID_RELATIONS)}", "type": "ValidationError"})

    graph_path.mkdir(parents=True)
    import yaml
    meta_data: dict = {
        "name": name,
        "doc_type": doc_type,
        "title": title,
        "authors": authors or [],
        "year": year,
        "venue": venue,
        "doi": doi,
        "abstract": abstract,
        "coverage": {"human": "none", "agent": "none"},
    }
    if date:
        meta_data["date"] = date
    if zotero_key:
        meta_data["zotero_key"] = zotero_key
    if content_hash:
        meta_data["content_hash"] = content_hash
    if source_path:
        meta_data["source_path"] = source_path
    if relations:
        meta_data["relations"] = relations

    meta = graph_path / "_meta.yaml"
    meta.write_text(yaml.dump(meta_data, default_flow_style=False), encoding="utf-8")

    logger.info("created source '%s' at %s", name, graph_path)
    with _graph_lock:
        graph = ZettelGraph(name)
        # On a REVIVED name, an outgoing instance may still hold its embeddings
        # lock with an in-flight lazy .kgl build. Inherit that lock (mirroring
        # _reloaded_graph) so both builds serialize on ONE lock instead of racing
        # the same disposable cache with a fresh lock.
        prev = _graphs.get(name)
        if prev is not None:
            graph._embeddings_lock = prev._embeddings_lock
        _graphs[name] = graph
    response: dict = {"name": name, "path": str(graph_path), "message": f"Created source '{name}'"}
    protocol = _extraction_protocol(doc_type)
    if "error" not in protocol:
        response["extraction_protocol"] = protocol
    return json.dumps(response)

analyze_gaps ¶

analyze_gaps(project: str = '', graph: str = '', limit: int = 25, gap_types: list[str] | None = None) -> str

Surface a TYPED, RANKED research agenda over the claim graph.

The successor to the old flat find_gaps: instead of five hard-coded lists, this is a typed scan that names WHY a region of the corpus is incomplete, scores HOW MUCH it matters (a bounded severity = salience × type-weight × actionability), and wires each finding to the producer-loop tool that would CLOSE it. Deterministic and LLM-free — every gap and score is reproducible from the graph. Gap TYPES:

evidential — a thin claim (few distinct supporting sources / low bounded strength) or a replication gap → expand_citations / claim('test').
dialectical — an unresolved debate (a contested, non-superseded claim) or an unarticulated counterclaim → claim('test').
structural — a foundational claim that is under-supported, a bridge claim (high centrality), or a frontier claim (no structural neighbours) → claim('test') / expand_citations.
coverage — a thinly-developed landscape hub → claim('propose').
comprehensiveness — a high-value unread (co-cited) citation not promoted to a local source → promote_citation.
temporal — a stale claim whose newest supporting paper lags the citation frontier → expand_citations (forward).

Plus six OPT-IN negative-space "opportunity" families (propose-only) that scan the MISSING edges + unmatched nodes across the graph and the memory↔ zettelkasten boundary. They are emitted ONLY when explicitly named in gap_types — the default scan is unchanged:

asymmetry — a practice memory-node with no aligned canon claim, or a canon claim with no aligned practice node (the cross-store overlay).
bridge — two dense clusters with (near-)zero connecting edges but high inter-cluster embedding similarity (a synthesis opportunity).
crux — an unresolved central debate (a contested camp, ranked by strength × centrality).
orphaned_question — a question note with no answering backlink.
void — a sparse-but-surrounded region of embedding space.
transfer — a method/model note that fits an untried adjacent cluster.

Each gap carries its type, structural predicate, severity, salience/type_weight/actionability components, the anchor it concerns, and the executable action (a producer-loop tool call).

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name to scope sources (ignored when `graph` is set).	`''`
`graph`	`str`	Single source graph to scope to (takes priority over `project`).	`''`
`limit`	`int`	Max ranked gaps to return (by severity desc).	`25`
`gap_types`	`list[str] \| None`	Optional subset of gap types to emit; empty = the six default claim-anchored types (the six opportunity families are opt-in).	`None`

Source code in zettelkasten/server.py

def analyze_gaps(
    project: str = "",
    graph: str = "",
    limit: int = 25,
    gap_types: list[str] | None = None,
) -> str:
    """Surface a TYPED, RANKED research agenda over the claim graph.

    The successor to the old flat ``find_gaps``: instead of five hard-coded
    lists, this is a typed scan that names WHY a region of the corpus is
    incomplete, scores HOW MUCH it matters (a bounded severity = salience ×
    type-weight × actionability), and wires each finding to the producer-loop
    tool that would CLOSE it. Deterministic and LLM-free — every gap and score is
    reproducible from the graph. Gap TYPES:

    * ``evidential`` — a thin claim (few distinct supporting sources / low bounded
      strength) or a replication gap → ``expand_citations`` / ``claim('test')``.
    * ``dialectical`` — an unresolved debate (a contested, non-superseded claim)
      or an unarticulated counterclaim → ``claim('test')``.
    * ``structural`` — a foundational claim that is under-supported, a bridge claim
      (high centrality), or a frontier claim (no structural neighbours) →
      ``claim('test')`` / ``expand_citations``.
    * ``coverage`` — a thinly-developed landscape hub → ``claim('propose')``.
    * ``comprehensiveness`` — a high-value unread (co-cited) citation not promoted
      to a local source → ``promote_citation``.
    * ``temporal`` — a stale claim whose newest supporting paper lags the citation
      frontier → ``expand_citations`` (forward).

    Plus six OPT-IN negative-space "opportunity" families (propose-only) that scan
    the MISSING edges + unmatched nodes across the graph and the memory↔
    zettelkasten boundary. They are emitted ONLY when explicitly named in
    ``gap_types`` — the default scan is unchanged:

    * ``asymmetry`` — a practice memory-node with no aligned canon claim, or a
      canon claim with no aligned practice node (the cross-store overlay).
    * ``bridge`` — two dense clusters with (near-)zero connecting edges but high
      inter-cluster embedding similarity (a synthesis opportunity).
    * ``crux`` — an unresolved central debate (a contested camp, ranked by
      strength × centrality).
    * ``orphaned_question`` — a ``question`` note with no answering backlink.
    * ``void`` — a sparse-but-surrounded region of embedding space.
    * ``transfer`` — a method/model note that fits an untried adjacent cluster.

    Each gap carries its ``type``, structural ``predicate``, ``severity``,
    ``salience``/``type_weight``/``actionability`` components, the ``anchor`` it
    concerns, and the executable ``action`` (a producer-loop tool call).

    Args:
        project: Project name to scope sources (ignored when ``graph`` is set).
        graph: Single source graph to scope to (takes priority over ``project``).
        limit: Max ranked gaps to return (by severity desc).
        gap_types: Optional subset of gap types to emit; empty = the six default
            claim-anchored types (the six opportunity families are opt-in).
    """
    from zettelkasten import claims

    wanted = gap_types or None
    # Per-type budgeting for the negative-space "opportunity" families auto-enables
    # inside ``analyze_gaps`` for whichever requested families are negative-space,
    # so an unbounded family (``asymmetry`` is O(#claims)) can never bury a rare one
    # (a lone ``crux``). The default claim-anchored scan keeps the single global slice.
    try:
        payload = claims.analyze_gaps(
            _get_graph,
            project=("" if graph else project),
            graph=graph,
            limit=limit if limit >= 0 else None,
            gap_types=wanted,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

rank_missing_papers ¶

rank_missing_papers(project: str = '', graph: str = '', limit: int = 25) -> str

Rank cited-but-unowned works as ACQUISITION targets.

The acquisition companion to analyze_gaps: where analyze_gaps names what is missing in the articulated graph, this names which WORKS to acquire to close it. A target is any cited-but-unowned work in the corpus (a citation with no local source folder), scored on a bounded 0–1 composite (claim-unlock + corpus influence + external citation mass). Claim-unlock leads: a work that is the CORE paper of an otherwise-uncovered key claim — uncovered because an unowned core cannot be read — ranks highest, since acquiring it is the only way to cover that claim. Deterministic and LLM-free; degenerate scopes (no claims, or no owned sources) are handled gracefully, not as errors.

Each target carries its score and components (influence, cited_by_count, unlocks_count + the unlocks claim list), a short reason, and the executable promote_citation action that acquires it.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name to scope sources (ignored when `graph` is set).	`''`
`graph`	`str`	Single source graph to scope to (takes priority over `project`).	`''`
`limit`	`int`	Max ranked targets to return (by score desc); negative = all.	`25`

Source code in zettelkasten/server.py

def rank_missing_papers(
    project: str = "",
    graph: str = "",
    limit: int = 25,
) -> str:
    """Rank cited-but-unowned works as ACQUISITION targets.

    The acquisition companion to ``analyze_gaps``: where ``analyze_gaps`` names
    what is missing in the articulated graph, this names which WORKS to acquire
    to close it. A target is any cited-but-unowned work in the corpus (a citation
    with no local source folder), scored on a bounded 0–1 composite (claim-unlock
    + corpus influence + external citation mass). Claim-unlock leads: a work that
    is the CORE paper of an otherwise-uncovered key claim — uncovered because an
    unowned core cannot be read — ranks highest, since acquiring it is the only
    way to cover that claim. Deterministic and LLM-free; degenerate scopes (no
    claims, or no owned sources) are handled gracefully, not as errors.

    Each target carries its ``score`` and components (``influence``,
    ``cited_by_count``, ``unlocks_count`` + the ``unlocks`` claim list), a short
    ``reason``, and the executable ``promote_citation`` ``action`` that acquires
    it.

    Args:
        project: Project name to scope sources (ignored when ``graph`` is set).
        graph: Single source graph to scope to (takes priority over ``project``).
        limit: Max ranked targets to return (by score desc); negative = all.
    """
    from zettelkasten import claims

    try:
        payload = claims.rank_missing_papers(
            _get_graph,
            project=("" if graph else project),
            graph=graph,
            limit=limit if limit >= 0 else None,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

build_syllabus ¶

build_syllabus(project: str = '', graph: str = '', budget: int = 12, suggested_only: bool = False, as_of: str = '') -> str

Assemble the Papers view: a scored, situated, theme-covered reading list.

The read-side companion to analyze_gaps: where analyze_gaps surfaces what's missing, this surfaces what to read and why. It runs the deterministic syllabus engine end-to-end (scoring → trends → interestingness/situate → theme model → set cover) over the corpus and returns one JSON payload the Papers view renders. No LLM, no network — every number is reproducible from the graph.

Scope is a single graph (one source), a project (its source set), or everything when both are empty. relation_edges and landscape hubs are derived from the graph's own links; embedding/cluster-driven signals (surprise, novelty, embedding-cluster themes) are left null here because the MCP server has no embedding machinery — the dashboard's /papers endpoint fills those in. Everything else (influence, salience, importance, anchor, read_priority, trajectory/velocity, dissent/brokerage/anomaly, situate labels, why_read/why_interesting, the suggested set) is fully populated.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name to scope sources (ignored when `graph` is set).	`''`
`graph`	`str`	Single source graph to scope to (takes priority over `project`).	`''`
`budget`	`int`	Max papers in the suggested reading set.	`12`
`suggested_only`	`bool`	Return only the works in the suggested set.	`False`
`as_of`	`str`	Replay trends as of this year (e.g. "2020"); empty = live.	`''`

Source code in zettelkasten/server.py

def build_syllabus(
    project: str = "",
    graph: str = "",
    budget: int = 12,
    suggested_only: bool = False,
    as_of: str = "",
) -> str:
    """Assemble the Papers view: a scored, situated, theme-covered reading list.

    The read-side companion to ``analyze_gaps``: where ``analyze_gaps`` surfaces
    what's missing, this surfaces what to *read* and why. It runs the deterministic
    syllabus engine end-to-end (scoring → trends → interestingness/situate →
    theme model → set cover) over the corpus and returns one JSON payload the
    Papers view renders. No LLM, no network — every number is reproducible from
    the graph.

    Scope is a single ``graph`` (one source), a ``project`` (its source set), or
    everything when both are empty. ``relation_edges`` and landscape hubs are
    derived from the graph's own links; embedding/cluster-driven signals
    (surprise, novelty, embedding-cluster themes) are left null here because the
    MCP server has no embedding machinery — the dashboard's ``/papers`` endpoint
    fills those in. Everything else (influence, salience, importance, anchor,
    read_priority, trajectory/velocity, dissent/brokerage/anomaly, situate
    labels, why_read/why_interesting, the suggested set) is fully populated.

    Args:
        project: Project name to scope sources (ignored when ``graph`` is set).
        graph: Single source graph to scope to (takes priority over ``project``).
        budget: Max papers in the suggested reading set.
        suggested_only: Return only the works in the suggested set.
        as_of: Replay trends as of this year (e.g. "2020"); empty = live.
    """
    from zettelkasten import papers

    # The HTTP /papers routes guard budget with Query(ge=0); mirror that contract
    # here so a negative budget through the MCP tool can't leak into the engine.
    budget = max(budget, 0)

    try:
        payload = papers.assemble_papers(
            _get_graph,
            project=project,
            graph=graph,
            budget=budget,
            suggested_only=suggested_only,
            as_of=as_of,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

build_claims ¶

build_claims(project: str = '', graph: str = '', limit: int = 0) -> str

Assemble the Claims view: ranked key claims with status and core papers.

The claim-level companion to build_syllabus: where the Papers view scores works, this scores the claims and findings the corpus asserts. It runs the deterministic, LLM-free claim engine (zettelkasten.claims) over the scoped graphs — extracting active claim/finding notes, building a one-pass reverse index of every support/corroboration/contestation/structural edge, and from that deriving each claim's bounded strength, derived status (established | contested | superseded), cross-source support span, and the single core paper to read for it. Returns one JSON payload: the scope, the claim and source counts, a status summary, and the ranked key_claims.

Scope is a single graph (one source), a project (its source set), or everything when both are empty. No LLM, no network, and no embeddings (mirrors build_syllabus); every score is reproducible from the graph.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name to scope sources (ignored when `graph` is set).	`''`
`graph`	`str`	Single source graph to scope to (takes priority over `project`).	`''`
`limit`	`int`	Max key claims to return; 0 (default) returns the full ranking.	`0`

Source code in zettelkasten/server.py

def build_claims(
    project: str = "",
    graph: str = "",
    limit: int = 0,
) -> str:
    """Assemble the Claims view: ranked key claims with status and core papers.

    The claim-level companion to ``build_syllabus``: where the Papers view scores
    *works*, this scores the *claims and findings* the corpus asserts. It runs the
    deterministic, LLM-free claim engine (``zettelkasten.claims``) over the scoped
    graphs — extracting active claim/finding notes, building a one-pass reverse
    index of every support/corroboration/contestation/structural edge, and from
    that deriving each claim's bounded strength, derived status (``established`` |
    ``contested`` | ``superseded``), cross-source support span, and the single
    core paper to read for it. Returns one JSON payload: the scope, the claim and
    source counts, a status summary, and the ranked ``key_claims``.

    Scope is a single ``graph`` (one source), a ``project`` (its source set), or
    everything when both are empty. No LLM, no network, and no embeddings (mirrors
    ``build_syllabus``); every score is reproducible from the graph.

    Args:
        project: Project name to scope sources (ignored when ``graph`` is set).
        graph: Single source graph to scope to (takes priority over ``project``).
        limit: Max key claims to return; 0 (default) returns the full ranking.
    """
    from zettelkasten import claims

    try:
        payload = claims.assemble_claims(
            _get_graph,
            project=("" if graph else project),
            graph=graph,
            limit=limit if limit > 0 else None,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

build_outline ¶

build_outline(project: str = '', graph: str = '', claim_ids: list[str] | None = None, as_of: str = '', name: str = '', outline_id: str = '', spine: str = '', spine_mode: str = '', ancestor_uids: list[str] | None = None, ancestor_levels: int = 0, force: bool = False, peek: bool = False) -> str

Build the grounded writing-scaffold outline for a scope (GATHER→DRAFT→write).

The writing-side companion to build_syllabus/build_claims: where those score the reading list and the claim ladder, this assembles the deterministic GATHER material for the scope, hands it to the DRAFT agent under the assemble-grounded-outline skill contract, runs the deterministic integrity post-pass, and persists the scaffold as a regenerable derived artifact at _reviews/<name>.md (the durable authored state stays the _reviews/<name>.yaml overlay). A matching generation signature with an extant artifact short-circuits to the cached markdown with no agent call (cached=True) unless force is set.

Scope is a single graph (one source), a project (its source set), or everything when both are empty. claim_ids pins the exact claims/order (else every active claim in salience order, with the claim-sparse fallback).

spine/ancestor_uids/ancestor_levels bring this tool to PARITY with the dashboard's POST /graphs|projects/{name}/outline route (which forwards the identical set to outline.build_outline). Omitting all three yields the historical flat, byte-identical single-spine/theme partition, so existing callers are unaffected.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name to scope sources (ignored when `graph` is set).	`''`
`graph`	`str`	Single source graph to scope to (takes priority over `project`).	`''`
`claim_ids`	`list[str] \| None`	Exact claim uids/ids to outline, in order; empty = all claims.	`None`
`as_of`	`str`	Replay trends as of this year (e.g. "2020"); empty = live.	`''`
`name`	`str`	Review name to persist under; defaults to graph/project/"all".	`''`
`outline_id`	`str`	Which of the review's outlines to build; empty = the manifest-backed default.	`''`
`spine`	`str`	Optional SPINE section partition (an organization id): the outline's sections become that spine's dimensions instead of the theme/settled partition. Empty = the historical theme partition.	`''`
`ancestor_uids`	`list[str] \| None`	AUTHORITATIVE nested-composition selection — the EXPLICIT, ordered (root→leaf) set of ancestor node uids to keep as framing tiers above the subject spine's leaf partition (the checked tiers of the frontend's projection). `None` = no explicit selection.	`None`
`ancestor_levels`	`int`	DEPRECATED count-based fallback for the nested selection, kept only for back-compat: `0` = today's flat partition, `N` = the nearest `N` ancestors, `-1` = every ancestor to the composition root. Prefer `ancestor_uids`, which takes PRECEDENCE whenever it is provided (`ancestor_levels` is consulted ONLY when `ancestor_uids` is omitted) — matching the route/projection exactly.	`0`
`force`	`bool`	Redraft even when the cached signature is unchanged.	`False`
`peek`	`bool`	Cost-free read-only staleness probe; never gathers/drafts/writes.	`False`

Source code in zettelkasten/server.py

def build_outline(
    project: str = "",
    graph: str = "",
    claim_ids: list[str] | None = None,
    as_of: str = "",
    name: str = "",
    outline_id: str = "",
    spine: str = "",
    spine_mode: str = "",
    ancestor_uids: list[str] | None = None,
    ancestor_levels: int = 0,
    force: bool = False,
    peek: bool = False,
) -> str:
    """Build the grounded writing-scaffold outline for a scope (GATHER→DRAFT→write).

    The writing-side companion to ``build_syllabus``/``build_claims``: where those
    score the reading list and the claim ladder, this assembles the deterministic
    GATHER material for the scope, hands it to the DRAFT agent under the
    ``assemble-grounded-outline`` skill contract, runs the deterministic integrity
    post-pass, and persists the scaffold as a regenerable derived artifact at
    ``_reviews/<name>.md`` (the durable authored state stays the
    ``_reviews/<name>.yaml`` overlay). A matching generation signature with an
    extant artifact short-circuits to the cached markdown with no agent call
    (``cached=True``) unless ``force`` is set.

    Scope is a single ``graph`` (one source), a ``project`` (its source set), or
    everything when both are empty. ``claim_ids`` pins the exact claims/order
    (else every active claim in salience order, with the claim-sparse fallback).

    ``spine``/``ancestor_uids``/``ancestor_levels`` bring this tool to PARITY with
    the dashboard's ``POST /graphs|projects/{name}/outline`` route (which forwards
    the identical set to ``outline.build_outline``). Omitting all three yields the
    historical flat, byte-identical single-spine/theme partition, so existing
    callers are unaffected.

    Args:
        project: Project name to scope sources (ignored when ``graph`` is set).
        graph: Single source graph to scope to (takes priority over ``project``).
        claim_ids: Exact claim uids/ids to outline, in order; empty = all claims.
        as_of: Replay trends as of this year (e.g. "2020"); empty = live.
        name: Review name to persist under; defaults to graph/project/"all".
        outline_id: Which of the review's outlines to build; empty = the
            manifest-backed default.
        spine: Optional SPINE section partition (an organization id): the outline's
            sections become that spine's dimensions instead of the theme/settled
            partition. Empty = the historical theme partition.
        ancestor_uids: AUTHORITATIVE nested-composition selection — the EXPLICIT,
            ordered (root→leaf) set of ancestor node uids to keep as framing tiers
            above the subject spine's leaf partition (the checked tiers of the
            frontend's projection). ``None`` = no explicit selection.
        ancestor_levels: DEPRECATED count-based fallback for the nested selection,
            kept only for back-compat: ``0`` = today's flat partition,
            ``N`` = the nearest ``N`` ancestors, ``-1`` = every ancestor to the
            composition root. Prefer ``ancestor_uids``, which takes PRECEDENCE
            whenever it is provided (``ancestor_levels`` is consulted ONLY when
            ``ancestor_uids`` is omitted) — matching the route/projection exactly.
        force: Redraft even when the cached signature is unchanged.
        peek: Cost-free read-only staleness probe; never gathers/drafts/writes.
    """
    from zettelkasten import outline

    try:
        payload = outline.build_outline(
            _get_graph,
            project=("" if graph else project),
            graph=graph,
            name=name,
            outline_id=outline_id,
            claim_ids=claim_ids,
            as_of=as_of,
            spine=spine,
            spine_mode=spine_mode,
            ancestor_levels=ancestor_levels,
            ancestor_uids=ancestor_uids,
            force=force,
            peek=peek,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

claim_anatomy ¶

claim_anatomy(graph: str, note_id: str, project: str = '') -> str

Assemble the full anatomy of a single claim or finding.

The claim-level detail companion to get_note: given a claim/finding's home graph and note_id, returns the claim (title/body/source locator), its evidence (supporting/replicating quote notes with page + grounding, strongest first), its strongest counterclaims with their own evidence, its caveats (qualifies) and any supersedes edges, the derived status, the bounded claim strength, the single :func:core_paper_atom to read, and an ordered reading path (prerequisite claims first). Read-only and LLM-free.

Claim ids are unique only WITHIN a graph, so both graph and note_id are required to identify the claim. By default the index is built over the home graph alone; pass project to widen the scope so cross-source supporters and counterclaims living in the project's other sources are seen. Returns {"error": ..., "type": "NotFound"} when no such claim exists in graph.

Parameters:

Name	Type	Description	Default
`graph`	`str`	The source graph the claim/finding lives in (its home graph).	required
`note_id`	`str`	The claim/finding note id (unique only within `graph`).	required
`project`	`str`	Optional project to widen the index scope for cross-source links.	`''`

Source code in zettelkasten/server.py

def claim_anatomy(
    graph: str,
    note_id: str,
    project: str = "",
) -> str:
    """Assemble the full anatomy of a single claim or finding.

    The claim-level detail companion to ``get_note``: given a claim/finding's home
    ``graph`` and ``note_id``, returns the claim (title/body/source locator), its
    evidence (supporting/replicating ``quote`` notes with page + grounding,
    strongest first), its strongest counterclaims *with their own evidence*, its
    caveats (``qualifies``) and any ``supersedes`` edges, the derived status, the
    bounded claim strength, the single :func:`core_paper_atom` to read, and an
    ordered reading path (prerequisite claims first). Read-only and LLM-free.

    Claim ids are unique only WITHIN a graph, so both ``graph`` and ``note_id`` are
    required to identify the claim. By default the index is built over the home
    ``graph`` alone; pass ``project`` to widen the scope so cross-source supporters
    and counterclaims living in the project's other sources are seen. Returns
    ``{"error": ..., "type": "NotFound"}`` when no such claim exists in ``graph``.

    Args:
        graph: The source graph the claim/finding lives in (its home graph).
        note_id: The claim/finding note id (unique only within ``graph``).
        project: Optional project to widen the index scope for cross-source links.
    """
    from zettelkasten import claims

    try:
        payload = claims.claim_anatomy(
            _get_graph,
            claim_id=note_id,
            source_graph=graph,
            project=project,
            graph="" if project else graph,
        )
    except Exception as e:
        return _json_tool_error(e)
    return json.dumps(payload)

syllabus ¶

syllabus(action: Literal['papers', 'claims', 'outline', 'gaps', 'missing', 'anatomy'], project: str = '', graph: str = '', note_id: str = '', budget: int = 12, limit: int = 0, suggested_only: bool = False, as_of: str = '', claim_ids: list[str] | None = None, name: str = '', outline_id: str = '', spine: str = '', spine_mode: str = '', ancestor_uids: list[str] | None = None, ancestor_levels: int = 0, force: bool = False, peek: bool = False, gap_types: list[str] | None = None) -> str

Use this when you need the read-only Workshop view: the reading list, claim ladder, writing outline, and typed research agenda over the corpus.

All actions are read-only, LLM-free, and reproducible from the graph. Scope is a single graph (one source), a project (its source set), or everything when both are empty.

Actions — name(required, optional?): papers(project?, graph?, budget?, suggested_only?, as_of?): scored, theme-covered reading list. claims(project?, graph?, limit?): ranked key claims with status and core papers (limit 0 = all). outline(project?, graph?, claim_ids?, as_of?, name?, outline_id?, spine?, spine_mode?, ancestor_uids?, ancestor_levels?, force?, peek?): grounded writing scaffold (dashboard-parity). spine partitions sections by an organization's dimensions; ancestor_uids selects nested-composition framing tiers (preferred over deprecated ancestor_levels); peek is a cost-free staleness probe; force redrafts. gaps(project?, graph?, limit?, gap_types?): typed, ranked research agenda over the claim graph (default limit 25). Six opt-in negative-space families (asymmetry/bridge/crux/orphaned_question/void/transfer) emit only when named in gap_types. missing(project?, graph?, limit?): cited-but-unowned works ranked as acquisition targets (default limit 25). anatomy(graph, note_id, project?): full anatomy of one claim/finding.

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def syllabus(
    action: Literal["papers", "claims", "outline", "gaps", "missing", "anatomy"],
    project: str = "",
    graph: str = "",
    note_id: str = "",
    budget: int = 12,
    limit: int = 0,
    suggested_only: bool = False,
    as_of: str = "",
    claim_ids: list[str] | None = None,
    name: str = "",
    outline_id: str = "",
    spine: str = "",
    spine_mode: str = "",
    ancestor_uids: list[str] | None = None,
    ancestor_levels: int = 0,
    force: bool = False,
    peek: bool = False,
    gap_types: list[str] | None = None,
) -> str:
    """Use this when you need the read-only Workshop view: the reading list, claim ladder, writing outline, and typed research agenda over the corpus.

    All actions are read-only, LLM-free, and reproducible from the graph. Scope
    is a single ``graph`` (one source), a ``project`` (its source set), or
    everything when both are empty.

    Actions — name(required, optional?):
        papers(project?, graph?, budget?, suggested_only?, as_of?): scored,
            theme-covered reading list.
        claims(project?, graph?, limit?): ranked key claims with status and core
            papers (``limit`` 0 = all).
        outline(project?, graph?, claim_ids?, as_of?, name?, outline_id?, spine?, spine_mode?, ancestor_uids?, ancestor_levels?, force?, peek?):
            grounded writing scaffold (dashboard-parity). ``spine`` partitions
            sections by an organization's dimensions; ``ancestor_uids`` selects
            nested-composition framing tiers (preferred over deprecated
            ``ancestor_levels``); ``peek`` is a cost-free staleness probe;
            ``force`` redrafts.
        gaps(project?, graph?, limit?, gap_types?): typed, ranked research agenda
            over the claim graph (default ``limit`` 25). Six opt-in negative-space
            families (asymmetry/bridge/crux/orphaned_question/void/transfer) emit
            only when named in ``gap_types``.
        missing(project?, graph?, limit?): cited-but-unowned works ranked as
            acquisition targets (default ``limit`` 25).
        anatomy(graph, note_id, project?): full anatomy of one claim/finding.

    All return JSON.
    """
    if action == "papers":
        return build_syllabus(
            project=project, graph=graph, budget=budget,
            suggested_only=suggested_only, as_of=as_of,
        )
    if action == "claims":
        return build_claims(project=project, graph=graph, limit=limit)
    if action == "outline":
        return build_outline(
            project=project, graph=graph, claim_ids=claim_ids, as_of=as_of,
            name=name, outline_id=outline_id, spine=spine, spine_mode=spine_mode,
            ancestor_uids=ancestor_uids, ancestor_levels=ancestor_levels,
            force=force, peek=peek,
        )
    if action == "gaps":
        return analyze_gaps(
            project=project, graph=graph,
            limit=(limit or 25), gap_types=gap_types,
        )
    if action == "missing":
        return rank_missing_papers(project=project, graph=graph, limit=(limit or 25))
    if action == "anatomy":
        if not graph or not note_id:
            return json.dumps({
                "error": "action='anatomy' requires graph and note_id.",
                "type": "BadRequest",
            })
        return claim_anatomy(graph=graph, note_id=note_id, project=project)
    return _bad_action(syllabus, action)

note ¶

note(action: Literal['get', 'search', 'follow', 'prerequisites', 'add', 'add_batch', 'update', 'link', 'attach', 'verify_grounding'], graph: str = '', note_id: str = '', title: str = '', query: str = '', type: str = '', tag: str = '', mode: str = 'keyword', top_k: int = 10, limit: int = 50, relation: str = '', body: str = '', source_book: str = '', source_chapter: str = '', source_page: str = '', tags: list[str] | None = None, links: list[dict] | None = None, notes: list[dict] | None = None, prerequisites: list[str] | None = None, aliases: list[str] | None = None, source_id: str = '', target_id: str = '', direction: str = 'both', equation: str = '', target_graph: str = '', append_body: str = '', new_type: str = '', add_tags: list[str] | None = None, add_links: list[dict] | None = None, add_aliases: list[str] | None = None, add_prerequisites: list[str] | None = None, project: str = '', synthesis_status: str = '', synthesis_graph: str = '', replace_body: str = '', data: dict | None = None, snapshot: dict | None = None, grounding: dict | None = None, epistemic_status: str = '', applies_when: dict | None = None) -> str

Use this when reading or writing notes: read (get/search/follow/prerequisites) or write (add/add_batch/update/link/attach/verify_grounding).

READ actions are always available; WRITE actions are refused at runtime when the server is write-free (ZK_DISABLE_WRITE). Every add/add_batch runs the same validation + verify_quote grounding gate.

Actions — name(required, optional?): get(graph, note_id?, title?): one note by id or title. search(graph, query?, type?, tag?, mode?, top_k?, limit?): keyword/semantic search or list (empty query lists notes). follow(graph, note_id, relation?, direction?, project?): links from a note (direction = outgoing|incoming|both). Pass project to also roll up CROSS-GRAPH incoming edges (a synthesis dimension's grounding). prerequisites(graph, note_id): the ordered prerequisite chain. add(graph, title, type, body, source_book?, source_chapter?, source_page?, tags?, links?, prerequisites?, aliases?, project?, synthesis_status?, synthesis_graph?, data?, snapshot?, grounding?, epistemic_status?, applies_when?): create a note. An equation note's body is LaTeX (render-checked) with an optional snapshot PDF crop; a figure note describes a chart/diagram in body with a snapshot image as its visual ground-truth (grounded when it resolves, else inferred); a claim may carry grounding + epistemic_status (grounded|measured|inferred); a requirement may carry applies_when (decision-tree, see guide). graph='_cross' + project auto-claims a synthesis note. add_batch(graph, notes, links?, project?, synthesis_graph?): create many notes in one box under a single lock + batched embed; a failure fails only that note. Returns {created, failed, links_added}. update(graph, note_id, append_body?, replace_body?, new_type?, add_tags?, add_links?, add_aliases?, add_prerequisites?, synthesis_status?): extend a note. body is an alias for replace_body here; synthesis_status flips a spine node scaffold→materialized. link(graph, source_id, target_id, relation, direction?, equation?, target_graph?): link two notes. attach(graph, note_id, tag, project?, synthesis_graph?): attach a claim to its schema dimension (tag) in a spine-backed run; the server resolves the dimension node, synthesis graph, relation, and edge direction from the persisted structure (the scribe never decides it). verify_grounding(graph, note_id): re-verify a method='data' grounding by RE-COMPUTING it; a clean reproduce stamps the note verified.

add/attach accept project + synthesis_graph to select the source's per-context extraction rubric. All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def note(
    action: Literal[
        "get", "search", "follow", "prerequisites", "add", "add_batch",
        "update", "link", "attach", "verify_grounding",
    ],
    graph: str = "",
    note_id: str = "",
    title: str = "",
    query: str = "",
    type: str = "",
    tag: str = "",
    mode: str = "keyword",
    top_k: int = 10,
    limit: int = 50,
    relation: str = "",
    body: str = "",
    source_book: str = "",
    source_chapter: str = "",
    source_page: str = "",
    tags: list[str] | None = None,
    links: list[dict] | None = None,
    notes: list[dict] | None = None,
    prerequisites: list[str] | None = None,
    aliases: list[str] | None = None,
    source_id: str = "",
    target_id: str = "",
    direction: str = "both",
    equation: str = "",
    target_graph: str = "",
    append_body: str = "",
    new_type: str = "",
    add_tags: list[str] | None = None,
    add_links: list[dict] | None = None,
    add_aliases: list[str] | None = None,
    add_prerequisites: list[str] | None = None,
    project: str = "",
    synthesis_status: str = "",
    synthesis_graph: str = "",
    replace_body: str = "",
    data: dict | None = None,
    snapshot: dict | None = None,
    grounding: dict | None = None,
    epistemic_status: str = "",
    applies_when: dict | None = None,
) -> str:
    """Use this when reading or writing notes: read (get/search/follow/prerequisites) or write (add/add_batch/update/link/attach/verify_grounding).

    READ actions are always available; WRITE actions are refused at runtime when
    the server is write-free (``ZK_DISABLE_WRITE``). Every ``add``/``add_batch``
    runs the same validation + ``verify_quote`` grounding gate.

    Actions — name(required, optional?):
        get(graph, note_id?, title?): one note by id or title.
        search(graph, query?, type?, tag?, mode?, top_k?, limit?): keyword/semantic
            search or list (empty ``query`` lists notes).
        follow(graph, note_id, relation?, direction?, project?): links from a note
            (``direction`` = outgoing|incoming|both). Pass ``project`` to also roll
            up CROSS-GRAPH incoming edges (a synthesis dimension's grounding).
        prerequisites(graph, note_id): the ordered prerequisite chain.
        add(graph, title, type, body, source_book?, source_chapter?, source_page?, tags?, links?, prerequisites?, aliases?, project?, synthesis_status?, synthesis_graph?, data?, snapshot?, grounding?, epistemic_status?, applies_when?):
            create a note. An ``equation`` note's ``body`` is LaTeX (render-checked)
            with an optional ``snapshot`` PDF crop; a ``figure`` note describes a
            chart/diagram in ``body`` with a ``snapshot`` image as its visual
            ground-truth (grounded when it resolves, else inferred); a claim may
            carry ``grounding`` + ``epistemic_status`` (grounded|measured|inferred);
            a ``requirement`` may carry ``applies_when`` (decision-tree, see ``guide``).
            ``graph='_cross'`` + ``project`` auto-claims a synthesis note.
        add_batch(graph, notes, links?, project?, synthesis_graph?): create many
            notes in one box under a single lock + batched embed; a failure fails
            only that note. Returns ``{created, failed, links_added}``.
        update(graph, note_id, append_body?, replace_body?, new_type?, add_tags?, add_links?, add_aliases?, add_prerequisites?, synthesis_status?):
            extend a note. ``body`` is an alias for ``replace_body`` here;
            ``synthesis_status`` flips a spine node scaffold→materialized.
        link(graph, source_id, target_id, relation, direction?, equation?, target_graph?):
            link two notes.
        attach(graph, note_id, tag, project?, synthesis_graph?): attach a claim to
            its schema dimension (``tag``) in a spine-backed run; the server
            resolves the dimension node, synthesis graph, relation, and edge
            direction from the persisted structure (the scribe never decides it).
        verify_grounding(graph, note_id): re-verify a ``method='data'`` grounding
            by RE-COMPUTING it; a clean reproduce stamps the note ``verified``.

    ``add``/``attach`` accept ``project`` + ``synthesis_graph`` to select the
    source's per-context extraction rubric. All return JSON.
    """
    if action == "get":
        return _as_json(get_note(graph=graph, note_id=note_id, title=title))
    if action == "search":
        return _as_json(search_notes(
            graph=graph, query=query, type=type, tag=tag,
            mode=mode, top_k=top_k, limit=limit,
        ))
    if action == "follow":
        return _as_json(follow_links(
            graph=graph, note_id=note_id, relation=relation,
            direction=direction or "both", project=project,
        ))
    if action == "prerequisites":
        return _as_json(get_prerequisites(graph=graph, note_id=note_id))
    if action == "add":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        projects = [project] if (graph == "_cross" and project) else []
        with _zk_write_lock(boxes=[graph], projects=projects):
            return _as_json(add_note(
                graph=graph, title=title, type=type, body=body,
                source_book=source_book, source_chapter=source_chapter,
                source_page=source_page, tags=tags, links=links,
                prerequisites=prerequisites, aliases=aliases, project=project,
                synthesis_status=synthesis_status, synthesis_graph=synthesis_graph,
                data=data, snapshot=snapshot,
                grounding=grounding, epistemic_status=epistemic_status,
                applies_when=applies_when,
            ))
    if action == "add_batch":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        # One box write lock covers the whole batch (and the ``_cross`` project
        # claim when authoring synthesis notes), mirroring the single-note ``add``.
        projects = [project] if (graph == "_cross" and project) else []
        with _zk_write_lock(boxes=[graph], projects=projects):
            return _as_json(add_notes_batch(
                graph=graph, notes=notes, links=links,
                project=project, synthesis_graph=synthesis_graph,
            ))
    if action == "update":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        # `body` is the `add` action's parameter, but agents reach for it on
        # `update` too, expecting a body rewrite. Without this it was silently
        # dropped (update only reads append_body/replace_body). Treat it as a
        # whole-body replace so the content isn't lost.
        if body:
            if replace_body or append_body:
                return _as_json({
                    "error": "Pass the new body via replace_body OR append_body, "
                             "not alongside body. For update, body is an alias "
                             "for replace_body.",
                    "type": "BadRequest",
                })
            replace_body = body
        with _zk_write_lock(boxes=[graph]):
            return _as_json(update_note(
                graph=graph, note_id=note_id, append_body=append_body,
                new_type=new_type, add_tags=add_tags, add_links=add_links,
                add_aliases=add_aliases, add_prerequisites=add_prerequisites,
                replace_body=replace_body, synthesis_status=synthesis_status,
            ))
    if action == "link":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        # ``direction`` defaults to "both" for the follow action; a link edge is
        # "outgoing" | "bidirectional", so map the shared default back here.
        link_direction = "outgoing" if direction == "both" else direction
        with _zk_write_lock(boxes=[graph, target_graph]):
            return _as_json(link_notes(
                graph=graph, source_id=source_id, target_id=target_id,
                relation=relation, direction=link_direction, equation=equation,
                target_graph=target_graph,
            ))
    if action == "attach":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        # Lock the source graph AND the synthesis graph the spine edge may touch
        # (resolved from the persisted structure) so concurrent scribes attaching
        # to the same spine cannot race on the synthesis-graph file. An explicit
        # context miss is not surfaced here (only used to widen the lock scope);
        # ``attach_to_dimension`` below re-resolves and returns the clean error.
        try:
            ext = resolve_extraction_context(
                graph, project=project or None, synthesis_graph=synthesis_graph or None,
            ) or {}
        except ExtractionContextNotFound:
            ext = {}
        structure = ext.get("structure") if isinstance(ext, dict) else None
        syn = str(structure.get("synthesis_graph") or "") if isinstance(structure, dict) else ""
        with _zk_write_lock(boxes=[graph, syn] if syn else [graph]):
            return _as_json(attach_to_dimension(
                graph=graph, note_id=note_id, tag=tag,
                project=project, synthesis_graph=synthesis_graph,
            ))
    if action == "verify_grounding":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[graph]):
            return _as_json(verify_grounding(graph=graph, note_id=note_id))
    return _bad_action(note, action)

source ¶

source(action: Literal['fulltext', 'outline', 'verify', 'coverage', 'protocol', 'ingest', 'create', 'set_coverage', 'backfill', 'mark_extracted'], source: str = '', doc_type: str = '', max_chars: int = 200000, name: str = '', title: str = '', authors: list[str] | None = None, year: int | None = None, venue: str = '', doi: str = '', abstract: str = '', zotero_key: str = '', content_hash: str = '', source_path: str = '', relations: list[dict] | None = None, key: str = '', path: str = '', agent: str = '', human: str = '', apply: bool = False, link_floor: float = 0.45, schema: str = '', project: str = '', synthesis_graph: str = '', offset: int = 0, page_start: int | None = None, page_end: int | None = None, cache_full: bool = False, graph: str = '', slices: list[dict] | None = None, run_id: str = '') -> str

Use this when reading or writing sources: read (fulltext/outline/verify/coverage/protocol) or write (ingest/create/set_coverage/backfill/mark_extracted).

READ actions are always available; WRITE actions are refused at runtime when write-free (ZK_DISABLE_WRITE).

Actions — name(required, optional?): fulltext(source, max_chars?, offset?, page_start?, page_end?): extracted full text; slice a book-scale source by offset (char start) or a 0-based page_start/page_end range. Reports total_chars/num_pages/sliced. outline(source): cached chapter outline + page index (empty when the source has no detectable bookmarks). verify(source): extraction QA report. coverage(source, schema?, project?, synthesis_graph?): deterministic Covered/Thin/Missing matrix over schema dimension tags (rubric from schema or the run's persisted per-context rubric). Advisory, model-independent. protocol(doc_type): the extraction protocol for a doc_type (no source). ingest(key?, path?, title?, max_chars?, cache_full?): resolve a paper to full text + a stable hash; provide exactly one of key (Zotero), path (local file), or title (Zotero lookup). create(name, doc_type, title?, authors?, year?, venue?, doi?, abstract?, zotero_key?, content_hash?, source_path?, relations?): create a source folder; dedups on content_hash. set_coverage(source, agent?, human?): set agent/human coverage flags. backfill(source, apply?, link_floor?): backfill grounded quotes for a source. mark_extracted(graph, slices, project?, synthesis_graph?, schema?, content_hash?, run_id?): record extracted slices (span dicts) onto the source's incremental ledger so the efficiency pipeline skips already-mined regions (graph = the source box).

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def source(
    action: Literal[
        "fulltext", "outline", "verify", "coverage", "protocol", "ingest",
        "create", "set_coverage", "backfill", "mark_extracted",
    ],
    source: str = "",
    doc_type: str = "",
    max_chars: int = 200_000,
    name: str = "",
    title: str = "",
    authors: list[str] | None = None,
    year: int | None = None,
    venue: str = "",
    doi: str = "",
    abstract: str = "",
    zotero_key: str = "",
    content_hash: str = "",
    source_path: str = "",
    relations: list[dict] | None = None,
    key: str = "",
    path: str = "",
    agent: str = "",
    human: str = "",
    apply: bool = False,
    link_floor: float = 0.45,
    schema: str = "",
    project: str = "",
    synthesis_graph: str = "",
    offset: int = 0,
    page_start: int | None = None,
    page_end: int | None = None,
    cache_full: bool = False,
    graph: str = "",
    slices: list[dict] | None = None,
    run_id: str = "",
) -> str:
    """Use this when reading or writing sources: read (fulltext/outline/verify/coverage/protocol) or write (ingest/create/set_coverage/backfill/mark_extracted).

    READ actions are always available; WRITE actions are refused at runtime when
    write-free (``ZK_DISABLE_WRITE``).

    Actions — name(required, optional?):
        fulltext(source, max_chars?, offset?, page_start?, page_end?): extracted
            full text; slice a book-scale source by ``offset`` (char start) or a
            0-based ``page_start``/``page_end`` range. Reports
            total_chars/num_pages/sliced.
        outline(source): cached chapter outline + page index (empty when the
            source has no detectable bookmarks).
        verify(source): extraction QA report.
        coverage(source, schema?, project?, synthesis_graph?): deterministic
            Covered/Thin/Missing matrix over schema dimension tags (rubric from
            ``schema`` or the run's persisted per-context rubric). Advisory,
            model-independent.
        protocol(doc_type): the extraction protocol for a doc_type (no source).
        ingest(key?, path?, title?, max_chars?, cache_full?): resolve a paper to
            full text + a stable hash; provide exactly one of ``key`` (Zotero),
            ``path`` (local file), or ``title`` (Zotero lookup).
        create(name, doc_type, title?, authors?, year?, venue?, doi?, abstract?, zotero_key?, content_hash?, source_path?, relations?):
            create a source folder; dedups on ``content_hash``.
        set_coverage(source, agent?, human?): set agent/human coverage flags.
        backfill(source, apply?, link_floor?): backfill grounded quotes for a source.
        mark_extracted(graph, slices, project?, synthesis_graph?, schema?, content_hash?, run_id?):
            record extracted ``slices`` (span dicts) onto the source's incremental
            ledger so the efficiency pipeline skips already-mined regions
            (``graph`` = the source box).

    All return JSON.
    """
    if action == "fulltext":
        return _as_json(get_fulltext(
            source=source, max_chars=max_chars,
            offset=offset, page_start=page_start, page_end=page_end,
        ))
    if action == "outline":
        return _as_json(get_source_outline(source=source))
    if action == "verify":
        return _as_json(verify_extraction(source=source))
    if action == "coverage":
        return _as_json(coverage_matrix(
            source=source, schema=schema,
            project=project, synthesis_graph=synthesis_graph))
    if action == "protocol":
        return _as_json(get_extraction_protocol(doc_type=doc_type))
    if action == "ingest":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        return _as_json(ingest_source(
            key=key, path=path, title=title,
            max_chars=max_chars, cache_full=cache_full,
        ))
    if action == "create":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[name]):
            return _as_json(create_source(
                name=name, doc_type=doc_type, title=title, authors=authors,
                year=year, venue=venue, doi=doi, abstract=abstract,
                zotero_key=zotero_key, content_hash=content_hash,
                source_path=source_path, relations=relations,
            ))
    if action == "set_coverage":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[source]):
            return _as_json(set_coverage(source=source, agent=agent, human=human))
    if action == "backfill":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[source]):
            return _as_json(backfill_quotes(source=source, apply=apply, link_floor=link_floor))
    if action == "mark_extracted":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        # ``graph`` names the source box per the tool contract; fall back to the
        # generic ``source``/``name`` identity args so either spelling works.
        src = graph or source or name
        from zettelkasten.graph_io import mark_slices_extracted

        with _zk_write_lock(boxes=[src]):
            try:
                record = mark_slices_extracted(
                    src, slices or [],
                    project=project or None, synthesis_graph=synthesis_graph or None,
                    schema=schema,
                    content_hash=content_hash, run_id=run_id, graphs_dir=GRAPHS_DIR,
                )
            except FileNotFoundError:
                return json.dumps({"error": f"Source '{src}' not found.", "type": "NotFound"})
            return _as_json(record)
    return _bad_action(source, action)

dataset ¶

dataset(action: Literal['add', 'get', 'values', 'snapshot', 'attach'], graph: str = '', note_id: str = '', title: str = '', body: str = '', backend: str = '', format: str = '', path: str = '', content_hash: str = '', schema: list[dict] | None = None, fetch: dict | None = None, tags: list[str] | None = None, links: list[dict] | None = None, source_book: str = '', source_chapter: str = '', source_page: str = '', project: str = '', target_id: str = '', target_graph: str = '', relation: str = 'measures', offset: int = 0, limit: int = 200) -> str

Use this when working with numerical/tabular datasets attached to a source: add/get/values/snapshot/attach.

A dataset is a typed note whose numbers live OUT of the markdown (a committed sidecar, a DVC-tracked file, or an API-fetched cache). READ actions are always available; WRITE actions (add/attach) are refused at runtime when write-free (ZK_DISABLE_WRITE).

Actions — name(required, optional?): add(graph, title, body?, backend?, format?, path?, content_hash?, schema?, fetch?, tags?, links?, source_book?, source_chapter?, source_page?, project?): create a dataset note with a storage pointer — backend = sidecar|dvc|api (inferred), format = csv|json|parquet (inferred from path), and either path (a file in the source folder) OR fetch ({provider, params}). get(graph, note_id): the note + resolved metadata (columns, row count, backend, warnings) WITHOUT row data. values(graph, note_id, offset?, limit?): resolve and return a paginated rows payload (limit default 200). snapshot(graph, note_id): re-read CURRENT source bytes and pin their hash as content_hash (reproducible offline reads / realigning a declared hash). Returns new + previous hash and whether it changed. attach(graph, note_id, target_id, target_graph?, relation?): link the dataset to a target (relation default measures; supports grounds a claim, derives-from for a derived series).

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def dataset(
    action: Literal["add", "get", "values", "snapshot", "attach"],
    graph: str = "",
    note_id: str = "",
    title: str = "",
    body: str = "",
    backend: str = "",
    format: str = "",
    path: str = "",
    content_hash: str = "",
    schema: list[dict] | None = None,
    fetch: dict | None = None,
    tags: list[str] | None = None,
    links: list[dict] | None = None,
    source_book: str = "",
    source_chapter: str = "",
    source_page: str = "",
    project: str = "",
    target_id: str = "",
    target_graph: str = "",
    relation: str = "measures",
    offset: int = 0,
    limit: int = 200,
) -> str:
    """Use this when working with numerical/tabular datasets attached to a source: add/get/values/snapshot/attach.

    A dataset is a typed note whose numbers live OUT of the markdown (a committed
    sidecar, a DVC-tracked file, or an API-fetched cache). READ actions are always
    available; WRITE actions (``add``/``attach``) are refused at runtime when
    write-free (``ZK_DISABLE_WRITE``).

    Actions — name(required, optional?):
        add(graph, title, body?, backend?, format?, path?, content_hash?, schema?, fetch?, tags?, links?, source_book?, source_chapter?, source_page?, project?):
            create a dataset note with a storage pointer — ``backend`` =
            sidecar|dvc|api (inferred), ``format`` = csv|json|parquet (inferred
            from ``path``), and either ``path`` (a file in the source folder) OR
            ``fetch`` ({provider, params}).
        get(graph, note_id): the note + resolved metadata (columns, row count,
            backend, warnings) WITHOUT row data.
        values(graph, note_id, offset?, limit?): resolve and return a paginated
            rows payload (``limit`` default 200).
        snapshot(graph, note_id): re-read CURRENT source bytes and pin their hash
            as ``content_hash`` (reproducible offline reads / realigning a declared
            hash). Returns new + previous hash and whether it changed.
        attach(graph, note_id, target_id, target_graph?, relation?): link the
            dataset to a target (``relation`` default ``measures``; ``supports``
            grounds a claim, ``derives-from`` for a derived series).

    All return JSON.
    """
    from zettelkasten import datasets as datasets_mod

    if action == "add":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        data_block = _build_data_block(
            backend=backend, data_format=format, path=path,
            content_hash=content_hash, schema=schema, fetch=fetch,
        )
        with _zk_write_lock(boxes=[graph]):
            return _as_json(add_note(
                graph=graph, title=title, type="dataset", body=body,
                source_book=source_book, source_chapter=source_chapter,
                source_page=source_page, tags=tags, links=links,
                project=project, data=data_block,
            ))
    if action == "get":
        loaded, err = _dataset_note(graph, note_id)
        if err:
            return _as_json(err)
        note_obj, graph_dir = loaded
        table = datasets_mod.resolve_dataset(note_obj, graph_dir)
        return _as_json({
            "id": note_obj.id,
            "title": note_obj.title,
            "graph": graph,
            "tags": note_obj.tags,
            "data": note_obj.data or {},
            "resolved": table.to_dict(include_rows=False),
        })
    if action == "values":
        loaded, err = _dataset_note(graph, note_id)
        if err:
            return _as_json(err)
        note_obj, graph_dir = loaded
        return _as_json(datasets_mod.read_dataset_values(
            note_obj, graph_dir, offset=offset, limit=limit,
        ))
    if action == "snapshot":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[graph]):
            loaded, err = _dataset_note(graph, note_id)
            if err:
                return _as_json(err)
            note_obj, graph_dir = loaded
            # Read the CURRENT source bytes (force past the fetch cache) and pin
            # their hash into the note. For an ``api``/``file`` dataset this also
            # (re)caches the bytes under that hash, so subsequent reads reproduce
            # this exact snapshot offline; for a path backend it just aligns the
            # declared hash with the file on disk (clearing any mismatch warning).
            table = datasets_mod.resolve_dataset(note_obj, graph_dir, force_live=True)
            if not table.actual_hash:
                return _as_json({
                    "error": (
                        "cannot snapshot: dataset resolved no bytes"
                        + (f" ({table.warning})" if table.warning else "")
                    ),
                    "type": "BadRequest",
                })
            data_block = dict(note_obj.data or {})
            previous = datasets_mod._normalize_hash(str(data_block.get("content_hash") or ""))
            data_block["content_hash"] = table.actual_hash
            note_obj.data = data_block
            zg = _get_graph(graph)
            zg.save_note(note_obj)
            return _as_json({
                "id": note_obj.id,
                "graph": graph,
                "content_hash": table.actual_hash,
                "previous_content_hash": previous,
                "changed": previous != table.actual_hash,
                "num_rows": table.num_rows,
                "backend": table.backend,
                "format": table.format,
            })
    if action == "attach":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[graph, target_graph]):
            return _as_json(link_notes(
                graph=graph, source_id=note_id, target_id=target_id,
                relation=relation or "measures", direction="outgoing",
                target_graph=target_graph,
            ))
    return _bad_action("dataset", action)

graph ¶

graph(action: Literal['list', 'summary', 'project', 'create', 'export', 'import'], graph: str = '', project: str = '', name: str = '', description: str = '', source: str = '', relations: list[dict] | None = None, format: str = 'gexf', source_dir: str = '', out_dir: str = '') -> str

Use this when managing zettelkasten graphs: read (list/summary/project), create, or export/import.

Reads always available; create/import are refused at runtime when write-free (ZK_DISABLE_WRITE); export is read-only.

Actions — name(required, optional?): list(): all concept boxes. summary(graph): summary of one box. project(project): render a project's combined graph. create(name, description?, source?, relations?): create a concept box. export(graph, format?, out_dir?): READ-ONLY export of one box. format="gexf" (default) returns a GEXF 1.2 document for Gephi/networkx; format="obsidian" writes one .md per note (frontmatter + Dataview wikilinks) under out_dir. import(graph, source_dir, format?): bulk-import an Obsidian/markdown vault (format="obsidian") into the box. Idempotent (dedup on content_hash); notes land as epistemic_status=inferred. Under ZK_PROPOSE_ONLY returns a dry-run manifest and writes nothing.

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def graph(
    action: Literal["list", "summary", "project", "create", "export", "import"],
    graph: str = "",
    project: str = "",
    name: str = "",
    description: str = "",
    source: str = "",
    relations: list[dict] | None = None,
    format: str = "gexf",
    source_dir: str = "",
    out_dir: str = "",
) -> str:
    """Use this when managing zettelkasten graphs: read (list/summary/project), create, or export/import.

    Reads always available; ``create``/``import`` are refused at runtime when
    write-free (``ZK_DISABLE_WRITE``); ``export`` is read-only.

    Actions — name(required, optional?):
        list(): all concept boxes.
        summary(graph): summary of one box.
        project(project): render a project's combined graph.
        create(name, description?, source?, relations?): create a concept box.
        export(graph, format?, out_dir?): READ-ONLY export of one box.
            ``format="gexf"`` (default) returns a GEXF 1.2 document for
            Gephi/networkx; ``format="obsidian"`` writes one ``.md`` per note
            (frontmatter + Dataview wikilinks) under ``out_dir``.
        import(graph, source_dir, format?): bulk-import an Obsidian/markdown vault
            (``format="obsidian"``) into the box. Idempotent (dedup on
            content_hash); notes land as ``epistemic_status=inferred``. Under
            ``ZK_PROPOSE_ONLY`` returns a dry-run manifest and writes nothing.

    All return JSON.
    """
    if action == "list":
        return _as_json(list_graphs())
    if action == "summary":
        return _as_json(get_graph_summary(graph=graph))
    if action == "project":
        return _as_json(project_graph(project=project))
    if action == "create":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=[name]):
            return _as_json(create_graph(
                name=name, description=description, source=source, relations=relations,
            ))
    if action == "export":
        return _as_json(export_graph(graph=graph, format=format, out_dir=out_dir))
    if action == "import":
        return _as_json(import_graph(graph=graph, source_dir=source_dir, format=format))
    return _bad_action("graph", action)

export_graph ¶

export_graph(graph: str, format: str = 'gexf', out_dir: str = '') -> dict

Export one box to GEXF or an Obsidian vault (read-only).

format="gexf" returns {format, graph, node_count, gexf} with the XML document inline; format="obsidian" writes one .md per note under out_dir (default <workspace>/export/<graph>-vault) and returns the write manifest.

Source code in zettelkasten/server.py

def export_graph(graph: str, format: str = "gexf", out_dir: str = "") -> dict:
    """Export one box to GEXF or an Obsidian vault (read-only).

    ``format="gexf"`` returns ``{format, graph, node_count, gexf}`` with the XML
    document inline; ``format="obsidian"`` writes one ``.md`` per note under
    ``out_dir`` (default ``<workspace>/export/<graph>-vault``) and returns the
    write manifest.
    """
    from zettelkasten import export as _export
    from zettelkasten.graph import _anchor

    try:
        _validate_name(graph)
    except ValueError as e:
        return {"error": str(e), "type": "ValidationError"}

    zg = _get_graph(graph)
    if format == "gexf":
        gexf = _export.graph_to_gexf(zg)
        return {
            "format": "gexf",
            "graph": graph,
            "node_count": len(zg.notes),
            "gexf": gexf,
        }
    if format == "obsidian":
        target = _anchor(Path(out_dir)) if out_dir else _anchor(
            Path("export") / f"{graph}-vault"
        )
        manifest = _export.graph_to_vault(zg, target)
        manifest.update({"format": "obsidian", "graph": graph})
        return manifest
    return {
        "error": f"Unknown export format '{format}'. Use gexf|obsidian.",
        "type": "BadRequest",
    }

import_graph ¶

import_graph(graph: str, source_dir: str, format: str = 'obsidian') -> dict

Import an Obsidian/markdown vault at source_dir into the graph box.

Honors ZK_PROPOSE_ONLY (dry-run manifest, no canonical writes) and the ZK_DISABLE_WRITE barrier (refused at runtime when not proposing).

Source code in zettelkasten/server.py

def import_graph(graph: str, source_dir: str, format: str = "obsidian") -> dict:
    """Import an Obsidian/markdown vault at ``source_dir`` into the ``graph`` box.

    Honors ``ZK_PROPOSE_ONLY`` (dry-run manifest, no canonical writes) and the
    ``ZK_DISABLE_WRITE`` barrier (refused at runtime when not proposing).
    """
    from zettelkasten import import_vault as _import_vault

    if format != "obsidian":
        return {
            "error": f"Unknown import format '{format}'. Use obsidian.",
            "type": "BadRequest",
        }
    try:
        _validate_name(graph)
    except ValueError as e:
        return {"error": str(e), "type": "ValidationError"}
    if not source_dir:
        return {"error": "source_dir is required for import.", "type": "BadRequest"}

    propose = _PROPOSE_ONLY
    if not propose:
        blocked = _writes_blocked()
        if blocked:
            return json.loads(blocked)

    try:
        if propose:
            zg = _get_graph(graph)
            return _import_vault.import_vault(source_dir, zg, propose_only=True)
        with _zk_write_lock(boxes=[graph]):
            zg = _get_graph(graph)
            return _import_vault.import_vault(source_dir, zg, propose_only=False)
    except ValueError as e:
        return {"error": str(e), "type": "ValidationError"}

citation ¶

citation(action: Literal['create', 'dedup', 'promote', 'expand_corpus', 'expand', 'unlink', 'export_bibliography', 'export'], id: str = '', title: str = '', authors: list[str] | None = None, year: int = 0, doc_type: str = '', doi: str = '', zotero_key: str = '', citation_id: str = '', dry_run: bool | None = None, source: str = '', sources: list[str] | None = None, remove_global: bool = False, direction: str = 'both', limit: int = 25, project: str = '', format: str = 'bibtex') -> str

Use this when managing citations, expanding a corpus, or exporting a bibliography.

WRITE actions mutate _citations and are refused at runtime when write-free; the read-only export_bibliography/export stays available even then.

Actions — name(required, optional?): create(id, title, authors, year, doc_type?, doi?, zotero_key?, source?): create a citation entity; source records it on that source's references list. dedup(dry_run?): dedup _citations (dry_run default true). promote(citation_id): promote a citation to a full local source. expand_corpus(citation_id?, zotero_key?): expand a citation's neighbourhood. expand(source?, doi?, direction?, limit?, dry_run?): expand a source's citation graph via OpenAlex (dry_run default false). unlink(citation_id, sources?, remove_global?, dry_run?): remove a citation's links (scoped to sources; remove_global=true also DELETEs the global entity, refused when deletes are off). export_bibliography / export (project?, format?): READ-ONLY bibliography from a project's sources/citations. format = bibtex (default) | formatted (author-year).

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def citation(
    action: Literal[
        "create", "dedup", "promote", "expand_corpus", "expand", "unlink",
        "export_bibliography", "export",
    ],
    id: str = "",
    title: str = "",
    authors: list[str] | None = None,
    year: int = 0,
    doc_type: str = "",
    doi: str = "",
    zotero_key: str = "",
    citation_id: str = "",
    dry_run: bool | None = None,
    source: str = "",
    sources: list[str] | None = None,
    remove_global: bool = False,
    direction: str = "both",
    limit: int = 25,
    project: str = "",
    format: str = "bibtex",
) -> str:
    """Use this when managing citations, expanding a corpus, or exporting a bibliography.

    WRITE actions mutate ``_citations`` and are refused at runtime when write-free;
    the read-only ``export_bibliography``/``export`` stays available even then.

    Actions — name(required, optional?):
        create(id, title, authors, year, doc_type?, doi?, zotero_key?, source?):
            create a citation entity; ``source`` records it on that source's
            ``references`` list.
        dedup(dry_run?): dedup ``_citations`` (``dry_run`` default true).
        promote(citation_id): promote a citation to a full local source.
        expand_corpus(citation_id?, zotero_key?): expand a citation's neighbourhood.
        expand(source?, doi?, direction?, limit?, dry_run?): expand a source's
            citation graph via OpenAlex (``dry_run`` default false).
        unlink(citation_id, sources?, remove_global?, dry_run?): remove a
            citation's links (scoped to ``sources``; ``remove_global=true`` also
            DELETEs the global entity, refused when deletes are off).
        export_bibliography / export (project?, format?): READ-ONLY bibliography
            from a project's sources/citations. ``format`` = bibtex (default) |
            formatted (author-year).

    All return JSON.
    """
    if action in ("export_bibliography", "export"):
        return _export_bibliography(project=project, format=format)
    if action == "unlink":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        if remove_global:
            blocked = _deletes_blocked()
            if blocked:
                return blocked
        with _zk_write_lock(boxes=["_citations", *(sources or [])]):
            return _as_json(unlink_citation(
                citation_id=citation_id or id,
                sources=sources,
                remove_global=remove_global,
                dry_run=(False if dry_run is None else dry_run),
            ))
    if action == "create":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=["_citations", source]):
            return _as_json(create_citation(
                id=id, title=title, authors=authors or [], year=year,
                doc_type=doc_type, doi=doi, zotero_key=zotero_key, source=source,
            ))
    if action == "dedup":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=["_citations"]):
            return _as_json(dedup_citations(dry_run=(True if dry_run is None else dry_run)))
    if action == "promote":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=["_citations"]):
            return _as_json(promote_citation(citation_id=citation_id))
    if action == "expand_corpus":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=["_citations"]):
            return _as_json(expand_corpus(citation_id=citation_id, zotero_key=zotero_key))
    if action == "expand":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        with _zk_write_lock(boxes=["_citations", source]):
            return _as_json(expand_citations(
                source=source, doi=doi, direction=direction, limit=limit,
                dry_run=(False if dry_run is None else dry_run),
            ))
    return _bad_action("citation", action)

project ¶

project(action: Literal['create', 'add', 'add_cross', 'remove_cross', 'set_default_spine'], name: str = '', description: str = '', sources: list[str] | None = None, cross: list[str] | None = None, project: str = '', source: str = '', cross_id: str = '', org_id: str = '') -> str

Use this to create or wire up projects (WRITE): create, add a source, cross-link a synthesis note, or set a default spine. Structurally unregistered when write-free.

Cross membership is EXPLICIT: a synthesis note belongs to a project iff its id is in the manifest's cross: list (one physical note in _cross/, many references). default_spine is the parallel registry for organizing STRUCTURES.

Actions — name(required, optional?): create(name, description?, sources?, cross?): create a project. add(project, source): add a source to a project. add_cross(project, cross_id): claim a _cross note for a project. remove_cross(project, cross_id): release a _cross note from a project. set_default_spine(project, org_id?): point a project's default_spine at org org_id; empty org_id clears it to the intrinsic lens.

All return JSON.

Source code in zettelkasten/server.py

@_write_tool()
def project(
    action: Literal["create", "add", "add_cross", "remove_cross", "set_default_spine"],
    name: str = "",
    description: str = "",
    sources: list[str] | None = None,
    cross: list[str] | None = None,
    project: str = "",
    source: str = "",
    cross_id: str = "",
    org_id: str = "",
) -> str:
    """Use this to create or wire up projects (WRITE): create, add a source, cross-link a synthesis note, or set a default spine. Structurally unregistered when write-free.

    Cross membership is EXPLICIT: a synthesis note belongs to a project iff its id
    is in the manifest's ``cross:`` list (one physical note in ``_cross/``, many
    references). ``default_spine`` is the parallel registry for organizing
    STRUCTURES.

    Actions — name(required, optional?):
        create(name, description?, sources?, cross?): create a project.
        add(project, source): add a source to a project.
        add_cross(project, cross_id): claim a ``_cross`` note for a project.
        remove_cross(project, cross_id): release a ``_cross`` note from a project.
        set_default_spine(project, org_id?): point a project's ``default_spine`` at
            org ``org_id``; empty ``org_id`` clears it to the intrinsic lens.

    All return JSON.
    """
    # NO ``project::P`` dispatch wrapper for the manifest mutators below. Every
    # _projects/<P>.yaml writer now serializes INTERNALLY on the project's OWNER
    # lock (``organizations._owner_lock_name('project', P)``) — the SAME lock
    # set_default_spine / delete_organization / add_cross's auto-claim
    # (_register_project_cross, reached from the note/claim write paths, which
    # bypass this dispatch) take. Locking internally (not here) is what unifies ALL
    # callers onto one lock; wrapping here too would just be a second, redundant
    # key and leave the bypassing callers unprotected.
    if action == "create":
        return _as_json(create_project(name=name, description=description, sources=sources, cross=cross))
    if action == "add":
        return _as_json(add_to_project(project=project, source=source))
    if action == "add_cross":
        return _as_json(add_cross_to_project(project=project, cross_id=cross_id))
    if action == "remove_cross":
        return _as_json(remove_cross_from_project(project=project, cross_id=cross_id))
    if action == "set_default_spine":
        # NO project::P wrapper here. ``set_default_spine_for_project`` serializes
        # INTERNALLY on the project's OWNER lock (``_owner_lock_name('project', P)``)
        # — the SAME non-reentrant lock ``delete_organization`` clears the pointer
        # under. Wrapping it again here would (a) double-acquire and deadlock if it
        # also took an owner-or-project lock, and (b) leave set/delete on different
        # locks and able to interleave into a dangling pointer. EXACTLY ONE lock
        # (the owner lock) is held, inside the function (defects A+B).
        return _as_json(set_default_spine_for_project(project=project, org_id=org_id))
    return _bad_action("project", action)

spine ¶

spine(action: Literal['list', 'promote', 'demote', 'resync', 'verify', 'delete'], org_id: str = '', project: str = '', graph: str = '', tag_stamp: bool = False, semantic: bool = False) -> str

Use this to manage a spine's lifecycle: list, promote, demote, resync, verify, or delete.

A spine definition (an owner-scoped arrangement of the corpus) is either a proposed lens (a read-only projection) or a materialized spine (a real synthesis graph); this tool drives the transitions. Every op except list targets one owner — pass project OR graph (not both). Set a project's primary "Organize by" spine via project(action="set_default_spine").

Actions — name(required, optional?): list(project): the spines/lenses in a project (read-only; same catalog as schema(action="spines")). promote(org_id, project?, graph?, tag_stamp?): materialize a proposed lens into a spine graph (apex + one dimension per column + one hub per row, bulk-attach members). tag_stamp (default OFF) also stamps each member's dimension tag onto its base note. WRITE. demote(org_id, project?, graph?): revert a spine to a lens while KEEPING its graph/nodes/edges (non-destructive). WRITE. resync(org_id, project?, graph?): incrementally route new/unrouted in-scope notes into an existing spine (idempotent). WRITE. verify(org_id, project?, graph?, semantic?): read-only synthesis audit flagging cells whose members are missing/ungrounded. semantic=true opts into a per-cell model call. delete(org_id, project?, graph?): revert the definition to a lens and delete its synthesis graph folder. Gated by the DELETE barrier. WRITE.

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def spine(
    action: Literal["list", "promote", "demote", "resync", "verify", "delete"],
    org_id: str = "",
    project: str = "",
    graph: str = "",
    tag_stamp: bool = False,
    semantic: bool = False,
) -> str:
    """Use this to manage a spine's lifecycle: list, promote, demote, resync, verify, or delete.

    A spine definition (an owner-scoped arrangement of the corpus) is either a
    proposed lens (a read-only projection) or a materialized spine (a real
    synthesis graph); this tool drives the transitions. Every op except ``list``
    targets one owner — pass ``project`` OR ``graph`` (not both). Set a project's
    primary "Organize by" spine via ``project(action="set_default_spine")``.

    Actions — name(required, optional?):
        list(project): the spines/lenses in a project (read-only; same catalog as
            schema(action="spines")).
        promote(org_id, project?, graph?, tag_stamp?): materialize a proposed lens
            into a spine graph (apex + one dimension per column + one hub per row,
            bulk-attach members). ``tag_stamp`` (default OFF) also stamps each
            member's dimension tag onto its base note. WRITE.
        demote(org_id, project?, graph?): revert a spine to a lens while KEEPING
            its graph/nodes/edges (non-destructive). WRITE.
        resync(org_id, project?, graph?): incrementally route new/unrouted in-scope
            notes into an existing spine (idempotent). WRITE.
        verify(org_id, project?, graph?, semantic?): read-only synthesis audit
            flagging cells whose members are missing/ungrounded. ``semantic=true``
            opts into a per-cell model call.
        delete(org_id, project?, graph?): revert the definition to a lens and
            delete its synthesis graph folder. Gated by the DELETE barrier. WRITE.

    All return JSON.
    """
    from zettelkasten import organizations as _organizations, tables

    if action == "list":
        proj, err = _resolve_project(project)
        if err:
            return json.dumps(err)
        return _as_json(discover_spines(project=proj, include_catalog=True))

    owner, err = _spine_owner_from_args(project, graph)
    if err is not None:
        return json.dumps(err)
    owner_type, owner_name = owner

    if action == "verify":
        try:
            return _as_json(_organizations.verify_organization(
                owner_type, owner_name, org_id, graphs_dir=GRAPHS_DIR,
                verify_fn=tables._default_synthesis_verify_fn if semantic else None,
            ))
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "NotFound"})

    if action == "delete":
        blocked = _deletes_blocked()
        if blocked is not None:
            return blocked
        try:
            result = _organizations.delete_spine(
                owner_type, owner_name, org_id, get_graph=_get_graph, graphs_dir=GRAPHS_DIR,
            )
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "NotFound"})
        # Evict the torn-down synthesis graph from the warm cache so a later read
        # never serves a now-deleted graph.
        deleted = result.get("deleted_graph") if isinstance(result, dict) else None
        if deleted:
            with _graph_lock:
                _graphs.pop(deleted, None)
                _graph_mtimes.pop(deleted, None)
        return _as_json(result)

    if action in ("promote", "demote", "resync"):
        blocked = _writes_blocked()
        if blocked is not None:
            return blocked
        try:
            if action == "promote":
                promoted = _organizations.promote_organization(
                    owner_type, owner_name, org_id,
                    get_graph=_get_graph, graphs_dir=GRAPHS_DIR,
                    group_fn=tables._default_extract_fn,
                    tag_stamp=bool(tag_stamp),
                )
                # Strong usage: promoting materializes these notes into a spine —
                # a deliberate synthesis action, so boost their access-recency.
                if isinstance(promoted, dict):
                    _record_notes_access(
                        [
                            (str(m.get("graph") or ""), str(m.get("note_id") or ""))
                            for m in promoted.get("attached_members") or []
                        ],
                        weight=3.0,
                    )
                return _as_json(promoted)
            if action == "demote":
                return _as_json(_organizations.demote_organization(
                    owner_type, owner_name, org_id, graphs_dir=GRAPHS_DIR,
                ))
            return _as_json(_organizations.resync_organization(
                owner_type, owner_name, org_id,
                get_graph=_get_graph, graphs_dir=GRAPHS_DIR,
                group_fn=tables._default_extract_fn,
            ))
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "NotFound"})

    return _bad_action("spine", action)

remine ¶

remine(action: Literal['propose', 'apply'], owner_type: str = '', owner_name: str = '', org_id: str = '', schema: str = '', intent: str = '', grid: dict | None = None, tag_stamp: bool = False, depth: int = 0) -> str

Use this when re-mining an org's corpus into its schema dimensions (propose → apply).

Two-phase, propose-first (the dashboard's approve-to-apply gate). Identify the org with owner_type (project|graph) + owner_name + org_id. apply is refused at runtime when write-free.

Actions — name(required, optional?): propose(owner_type, owner_name, org_id, schema?, intent?, depth?): READ-ONLY. Classify the org's row-axis scope into a matrix-shaped grid (cells carry inferred members + a residual bucket); writes nothing. An intent ("make a matrix for X") INDUCES columns from the corpus (flagged induced=True) instead of the org's fixed dimensions; depth>=1 (default 0) sub-clusters each induced column into children (emergent columns only). apply(owner_type, owner_name, org_id, grid, tag_stamp?): promote the approved grid (the propose payload) via the attach seam, writing spine-member edges. tag_stamp (default OFF) also stamps each dimension tag onto members' base notes.

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def remine(
    action: Literal["propose", "apply"],
    owner_type: str = "",
    owner_name: str = "",
    org_id: str = "",
    schema: str = "",
    intent: str = "",
    grid: dict | None = None,
    tag_stamp: bool = False,
    depth: int = 0,
) -> str:
    """Use this when re-mining an org's corpus into its schema dimensions (propose → apply).

    Two-phase, propose-first (the dashboard's approve-to-apply gate). Identify the
    org with ``owner_type`` (project|graph) + ``owner_name`` + ``org_id``.
    ``apply`` is refused at runtime when write-free.

    Actions — name(required, optional?):
        propose(owner_type, owner_name, org_id, schema?, intent?, depth?):
            READ-ONLY. Classify the org's row-axis scope into a matrix-shaped grid
            (cells carry inferred members + a ``residual`` bucket); writes nothing.
            An ``intent`` ("make a matrix for X") INDUCES columns from the corpus
            (flagged ``induced=True``) instead of the org's fixed dimensions;
            ``depth>=1`` (default 0) sub-clusters each induced column into
            ``children`` (emergent columns only).
        apply(owner_type, owner_name, org_id, grid, tag_stamp?): promote the
            approved ``grid`` (the propose payload) via the attach seam, writing
            ``spine-member`` edges. ``tag_stamp`` (default OFF) also stamps each
            dimension tag onto members' base notes.

    All return JSON.
    """
    if action == "propose":
        return _as_json(_remine_propose(owner_type, owner_name, org_id, schema=schema, intent=intent, depth=depth))
    if action == "apply":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        return _as_json(_remine_apply(owner_type, owner_name, org_id, grid=grid, tag_stamp=tag_stamp))
    return _bad_action("remine", action)

discover_spines ¶

discover_spines(project: str = '', *, include_catalog: bool = True) -> dict

Find materialized synthesis spines living in a project.

A spine is an apex node (tagged synthesis + extraction-synthesis) with materialized dimension nodes (tagged extraction-dimension) rolling up into it, plus an optional spec stub, in a dedicated synthesis graph. This scans the project the caller is in and returns each spine as a concrete shape a schema author can model or extend — so a new schema's synthesis block can mirror the apex type and dimension tags that already exist, instead of leaning on a shipped example.

Spines live in TWO places, and a complete answer must look at both:

A LEGACY extraction spine (coordinator.extraction.prep_spine) is registered as a project SOURCE — found by the sources scan.
A PROMOTED / PORTED spine (organizations.promote_organization or the lazy legacy port) is deliberately NOT a project source; it lives in the org registry as a state=='spine' org naming its synthesis graph in spine_ref (the spine CATALOG, per the "spines are additive/selectable layers, NOT project sources" decision). Under the current org model this is the majority case — a project whose spines are all promoted used to read back EMPTY here because only sources were scanned.

include_catalog (default True) unions the catalog's spine graphs into the scan so BOTH kinds surface; a promoted/ported spine also carries its org_id/ org_title in the result. It is set False by the port_spines caller (which injects this as its discover_fn): porting must see only source-registered legacy spines, since stamping a promoted spine into the ported-marker would wrongly re-register its graph as a source on a later delete.

project may be omitted when the repo has exactly one project (the common case); otherwise the available project names are returned so the caller can disambiguate.

Source code in zettelkasten/server.py

def discover_spines(project: str = "", *, include_catalog: bool = True) -> dict:
    """Find materialized synthesis spines living in a project.

    A *spine* is an apex node (tagged ``synthesis`` + ``extraction-synthesis``)
    with materialized dimension nodes (tagged ``extraction-dimension``) rolling up
    into it, plus an optional spec stub, in a dedicated synthesis graph. This scans
    the project the caller is in and returns each spine as a concrete shape a schema
    author can model or extend — so a new schema's ``synthesis`` block can mirror
    the apex type and dimension tags that already exist, instead of leaning on a
    shipped example.

    Spines live in TWO places, and a complete answer must look at both:

    * A LEGACY extraction spine (``coordinator.extraction.prep_spine``) is
      registered as a project SOURCE — found by the sources scan.
    * A PROMOTED / PORTED spine (``organizations.promote_organization`` or the lazy
      legacy port) is deliberately NOT a project source; it lives in the org
      registry as a ``state=='spine'`` org naming its synthesis graph in
      ``spine_ref`` (the spine CATALOG, per the "spines are additive/selectable
      layers, NOT project sources" decision). Under the current org model this is
      the majority case — a project whose spines are all promoted used to read back
      EMPTY here because only sources were scanned.

    ``include_catalog`` (default True) unions the catalog's spine graphs into the
    scan so BOTH kinds surface; a promoted/ported spine also carries its ``org_id``/
    ``org_title`` in the result. It is set False by the ``port_spines`` caller (which
    injects this as its ``discover_fn``): porting must see only source-registered
    legacy spines, since stamping a promoted spine into the ported-marker would
    wrongly re-register its graph as a source on a later delete.

    ``project`` may be omitted when the repo has exactly one project (the common
    case); otherwise the available project names are returned so the caller can
    disambiguate.
    """
    from zettelkasten.graph import list_projects, load_project
    from zettelkasten.spine import APEX_TAGS, DIMENSION_TAG, SPEC_TAG

    proj = (project or "").strip()
    if not proj:
        names = [p.get("name") for p in list_projects() if p.get("name")]
        if len(names) == 1:
            proj = names[0]
        elif not names:
            return {"error": "No projects exist yet.", "type": "NotFound"}
        else:
            return {
                "error": "Multiple projects exist; pass project= to pick one.",
                "projects": names,
                "type": "AmbiguousProject",
            }

    data = load_project(proj)
    if not data:
        return {"error": f"Project '{proj}' not found.", "type": "NotFound"}

    # Read the org registry ONCE to (a) collect the spine CATALOG — every
    # ``state=='spine'`` org's ``spine_ref`` graph, the promoted/ported spines that
    # are NOT project sources — and (b) exclude DEMOTED (``lens``-state) orgs' spine
    # graphs. A demoted org keeps its apex graph + nodes (so a re-promote still
    # finds them) but NULLs ``spine_ref`` and stashes the preserved graph in
    # ``last_spine_ref`` (the W4 single-source-of-truth contract), so the graph to
    # exclude comes from ``last_spine_ref`` for a demoted org (with a ``spine_ref``
    # fallback for any legacy lens that still carries one). ``migrate=False`` is
    # REQUIRED: a migrate=True listing lazily fires ``port_spines``, which calls back
    # into this function (infinite recursion). Skipping migration is safe — an
    # un-ported legacy spine is still a project source and so is covered by the
    # sources term below.
    demoted_graphs: set[str] = set()
    catalog_graphs: set[str] = set()
    org_by_graph: dict[str, dict] = {}
    try:
        from zettelkasten import organizations as _organizations

        for _org in _organizations.list_organizations("project", proj, migrate=False):
            ref = (_org.get("spine_ref") or "").strip()
            if _org.get("state") == "spine":
                if not ref:
                    continue
                catalog_graphs.add(ref)
                org_by_graph.setdefault(
                    ref,
                    {"org_id": _org.get("id") or "", "org_title": _org.get("title") or ""},
                )
            else:
                # A demoted org's preserved graph lives in ``last_spine_ref`` now;
                # fall back to ``spine_ref`` for a legacy lens that still has one.
                demoted_ref = ref or (_org.get("last_spine_ref") or "").strip()
                if demoted_ref:
                    demoted_graphs.add(demoted_ref)
    except Exception:  # noqa: BLE001 — discovery degrades gracefully without the registry
        pass

    # Scan the project's source graphs first (stable, deterministic ordering), then
    # the catalog spine graphs when requested; ``dict.fromkeys`` dedups a graph that
    # is both a source and a catalog entry. Demoted graphs are dropped from either.
    candidate_graphs = list(data.get("sources", []) or [])
    if include_catalog:
        candidate_graphs.extend(sorted(catalog_graphs))
    scan_graphs = [g for g in dict.fromkeys(candidate_graphs) if g not in demoted_graphs]

    spines: list[dict] = []
    for gname in scan_graphs:
        try:
            zg = _get_graph(gname)
        except Exception:
            continue
        for note in zg.notes.values():
            tags = set(note.tags or [])
            if not (set(APEX_TAGS) <= tags):
                continue  # not a prep-built apex
            apex_id = note.id
            dimensions: list[dict] = []
            spec: dict | None = None
            for back in zg.get_backlinks(apex_id):
                src = zg.notes.get(back["source_id"])
                if not src:
                    continue
                src_tags = set(src.tags or [])
                entry = {
                    "id": src.id,
                    "title": src.title,
                    "type": src.type,
                    "relation": back["relation"],
                    "synthesis_status": src.synthesis_status or "",
                }
                if DIMENSION_TAG in src_tags:
                    entry["tag"] = next(
                        (t for t in (src.tags or []) if t != DIMENSION_TAG),
                        "",
                    )
                    dimensions.append(entry)
                elif SPEC_TAG in src_tags:
                    spec = entry
            synthesizes = [
                {
                    "target_id": link["target_id"],
                    "target_title": link.get("target_title"),
                    "graph": link.get("graph", gname),
                    "relation": link["relation"],
                }
                for link in zg.get_linked(apex_id)
            ]
            # Surface synthesis lifecycle so "which dimensions/personas are
            # unfinished?" is a one-query answer (apex + any dimension still
            # tagged scaffold), not a body-char-count forensic.
            incomplete = [
                d["title"] for d in dimensions
                if (d.get("synthesis_status") or "") == "scaffold"
            ]
            apex_status = note.synthesis_status or ""
            if apex_status == "scaffold":
                incomplete.append(note.title)
            entry = {
                "graph": gname,
                "apex": {
                    "id": apex_id,
                    "title": note.title,
                    "type": note.type,
                    "synthesis_status": apex_status,
                },
                "spec": spec,
                "dimensions": sorted(
                    dimensions, key=lambda d: d.get("tag") or d["title"]
                ),
                "synthesizes": synthesizes,
                "incomplete": sorted(incomplete),
                "materialized": not incomplete,
            }
            # A catalog (promoted/ported) spine also carries its owning org's id so
            # a caller can act on it (overlay, set as default, resync) without a
            # second registry lookup. Legacy source spines carry no org and omit it.
            org_meta = org_by_graph.get(gname)
            if org_meta and org_meta.get("org_id"):
                entry["org_id"] = org_meta["org_id"]
                entry["org_title"] = org_meta["org_title"]
            spines.append(entry)

    return {
        "project": proj,
        "count": len(spines),
        "spines": spines,
        "note": (
            "No materialized spines found in this project's graphs — author a "
            "schema with a synthesis block to create one."
            if not spines
            else "Each spine is a live example: model a new schema's synthesis "
            "block (apex type + dimension tags) on the structures below."
        ),
    }

schema ¶

schema(action: Literal['list', 'get', 'spines', 'validate', 'save'], name: str = '', spec: dict | None = None, project: str = '') -> str

Use this when working with extraction schemas — the rubric a grounded-extraction run reads documents through: list/get/spines/validate/save.

A schema is a run's FOCUS LENS: named dimensions (each tag becomes a note tag + cross-source comparison index), an optional relations backbone, and an optional synthesis block that materializes a persistent spine (apex + one node per dimension) in a dedicated graph. You supply the judgment (drafting from a goal, optionally after reading a sample via source(action="fulltext")); this tool guarantees a valid block lands in the registry. Author TWO deliberate policies: EXTRACTION (how claims come out — dimensions + grounded/strict + hub) and SYNTHESIS (how they roll up — a synthesis block, or a deliberate flat checklist; "no block" must be a choice, not an omission). A declared synthesis spine is DONE only once its apex/dimension prose is authored by hand (prep pre-creates empty nodes).

Actions — name(required, optional?): list(): names of all registered schemas (package defaults + project). get(name): the fully-expanded schema object. spines(project?): discover materialized synthesis spines already living in the project — legacy source spines AND promoted/ported catalog spines (which carry their org_id/org_title). Call before authoring so a new synthesis block mirrors existing structure. validate(spec): expand a draft spec (raw schema block) WITHOUT writing; returns the expanded object or a validation error. save(name, spec): validate then write spec under name into the project's extraction-schemas.yaml (overrides a built-in of the same name). WRITE; refused when write-free. Restart the coordinator + zettelkasten MCP servers to pick it up (registry cached at load).

A spec is the raw YAML block as a dict, e.g.::

{"description": "...", "grounded": true, "link_relation": "input-to",
 "hub": {"type": "concept", "title_template": "{persona}: Profile"},
 "dimensions": [{"tag": "attends-to", "desc": "..."}, ...],
 "relations": [{"from": "ranking-rule", "to": "attends-to",
                 "relation": "depends-on"}],
 "synthesis": {"graph": "{label}-profile",
                "node": {"type": "synthesis", "title_template": "{label}: Profile",
                          "link_relation": "related"},
                "dimension_node": {"type": "concept",
                                    "title_template": "{label} - {dimension}",
                                    "attach_relation": "input-to",
                                    "relation": "component-of"}}}

Source code in zettelkasten/server.py

@mcp_server.tool()
def schema(
    action: Literal["list", "get", "spines", "validate", "save"],
    name: str = "",
    spec: dict | None = None,
    project: str = "",
) -> str:
    """Use this when working with extraction schemas — the rubric a grounded-extraction run reads documents through: list/get/spines/validate/save.

    A schema is a run's FOCUS LENS: named ``dimensions`` (each ``tag`` becomes a
    note tag + cross-source comparison index), an optional ``relations`` backbone,
    and an optional ``synthesis`` block that materializes a persistent spine (apex
    + one node per dimension) in a dedicated graph. You supply the judgment
    (drafting from a goal, optionally after reading a sample via
    ``source(action="fulltext")``); this tool guarantees a valid block lands in the
    registry. Author TWO deliberate policies: EXTRACTION (how claims come out —
    ``dimensions`` + ``grounded``/``strict`` + ``hub``) and SYNTHESIS (how they
    roll up — a ``synthesis`` block, or a deliberate flat checklist; "no block"
    must be a choice, not an omission). A declared synthesis spine is DONE only
    once its apex/dimension prose is authored by hand (prep pre-creates empty nodes).

    Actions — name(required, optional?):
        list(): names of all registered schemas (package defaults + project).
        get(name): the fully-expanded schema object.
        spines(project?): discover materialized synthesis spines already living in
            the project — legacy source spines AND promoted/ported catalog spines
            (which carry their ``org_id``/``org_title``). Call before authoring so a
            new ``synthesis`` block mirrors existing structure.
        validate(spec): expand a draft ``spec`` (raw schema block) WITHOUT writing;
            returns the expanded object or a validation error.
        save(name, spec): validate then write ``spec`` under ``name`` into the
            project's ``extraction-schemas.yaml`` (overrides a built-in of the same
            name). WRITE; refused when write-free. Restart the coordinator +
            zettelkasten MCP servers to pick it up (registry cached at load).

    A ``spec`` is the raw YAML block as a dict, e.g.::

        {"description": "...", "grounded": true, "link_relation": "input-to",
         "hub": {"type": "concept", "title_template": "{persona}: Profile"},
         "dimensions": [{"tag": "attends-to", "desc": "..."}, ...],
         "relations": [{"from": "ranking-rule", "to": "attends-to",
                         "relation": "depends-on"}],
         "synthesis": {"graph": "{label}-profile",
                        "node": {"type": "synthesis", "title_template": "{label}: Profile",
                                  "link_relation": "related"},
                        "dimension_node": {"type": "concept",
                                            "title_template": "{label} - {dimension}",
                                            "attach_relation": "input-to",
                                            "relation": "component-of"}}}
    """
    from zettelkasten import extraction_schemas

    if action == "list":
        return _as_json({"schemas": extraction_schemas.list_schemas()})
    if action == "spines":
        return _as_json(discover_spines(project=project, include_catalog=True))
    if action == "get":
        obj = extraction_schemas.expand_schema(name)
        if obj is None:
            return json.dumps({
                "error": (
                    f"Unknown schema '{name}'. Available: "
                    f"{extraction_schemas.list_schemas()}"
                ),
                "type": "NotFound",
            })
        return _as_json(obj)
    if action == "validate":
        if not isinstance(spec, dict):
            return json.dumps({
                "error": "validate requires a 'spec' mapping.", "type": "BadRequest",
            })
        try:
            expanded = extraction_schemas.expand_spec(name or "(draft)", spec)
        except ValueError as e:
            return json.dumps({"valid": False, "error": str(e), "type": "ValidationError"})
        return _as_json({"valid": True, "schema": expanded})
    if action == "save":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        if not isinstance(spec, dict):
            return json.dumps({
                "error": "save requires a 'spec' mapping.", "type": "BadRequest",
            })
        try:
            result = extraction_schemas.save_schema(name, spec)
        except ValueError as e:
            return json.dumps({"error": str(e), "type": "ValidationError"})
        result["note"] = (
            "Schema saved. Restart the coordinator + zettelkasten MCP servers "
            "(Cursor Settings > MCP) so the cached registry reloads."
        )
        return _as_json(result)
    return _bad_action("schema", action)

zotero ¶

zotero(action: Literal['lookup', 'collections', 'list', 'fetch_pdf'], query: str = '', key: str = '', collection: str = '', limit: int = 200, include_text: bool = True, include_annotations: bool = True) -> str

Use this when you need read-only Zotero library access: lookup, collections, list, or fetch a PDF.

Actions — name(required, optional?): lookup(query): search the library by title/author/keyword. collections(): list collections (folders) with key, name, path, parent, and item count — browse what the library actually holds. list(collection?, limit?): papers in a collection (key preferred, or name); empty collection lists the most recently added top-level items (limit default 200). Same summaries as lookup, ready for fetch_pdf. fetch_pdf(key, include_text?, include_annotations?): resolve a Zotero item to its local PDF (full text + annotations).

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def zotero(
    action: Literal["lookup", "collections", "list", "fetch_pdf"],
    query: str = "",
    key: str = "",
    collection: str = "",
    limit: int = 200,
    include_text: bool = True,
    include_annotations: bool = True,
) -> str:
    """Use this when you need read-only Zotero library access: lookup, collections, list, or fetch a PDF.

    Actions — name(required, optional?):
        lookup(query): search the library by title/author/keyword.
        collections(): list collections (folders) with key, name, path, parent, and
            item count — browse what the library actually holds.
        list(collection?, limit?): papers in a collection (key preferred, or name);
            empty ``collection`` lists the most recently added top-level items
            (``limit`` default 200). Same summaries as ``lookup``, ready for
            ``fetch_pdf``.
        fetch_pdf(key, include_text?, include_annotations?): resolve a Zotero item
            to its local PDF (full text + annotations).

    All return JSON.
    """
    if action == "lookup":
        return _as_json(zotero_lookup(query=query))
    if action == "collections":
        return _as_json(zotero_collections())
    if action == "list":
        return _as_json(zotero_list_items(collection=collection, limit=limit))
    if action == "fetch_pdf":
        return _as_json(zotero_fetch_pdf(
            key=key, include_text=include_text, include_annotations=include_annotations,
        ))
    return _bad_action("zotero", action)

review ¶

review(action: Literal['list', 'load', 'create', 'update', 'rename', 'delete', 'suggest_claims', 'propose_placements'], name: str = '', *, new_name: str = '', title: str = '', project: str = '', graph: str = '', question: str = '', ordering: list | None = None, set_tier: dict | None = None, exclude: list[str] | None = None, unexclude: list[str] | None = None, add_scaffold: dict | None = None, remove_scaffold: list[str] | None = None, rename_section: dict | None = None, set_section_note: dict | None = None, set_placement: dict | None = None, favorite: list[str] | None = None, unfavorite: list[str] | None = None, add_proposed_claim: dict | None = None, remove_proposed_claim: list[str] | None = None, add_proposed_theme: dict | None = None, remove_proposed_theme: list[str] | None = None, remove_derived_theme: list[str] | None = None, restore_derived_theme: list[str] | None = None, text: str = '', top_k: int = 0, granularity: int = 50, whole_pool: bool = False) -> str

Use this when reading or authoring a literature-review overlay (the _reviews/ editorial layer).

A review is a curated, theme-structured narrative over the corpus with two separate halves: the PROJECTION (a deterministic, LLM-free view of the corpus into themes + a top-level unplaced worklist, recomputed on every read) and the OVERLAY (the author's durable editorial decisions stored as ID-references — the SOURCE OF TRUTH; load reconciles it onto a fresh projection). Scope is LOCAL-REPO ONLY; writes are crash-safe (per-review lock, atomic rename, debounced zettelkasten-mcp commit, path-traversal-validated name).

Actions — name(required, optional?): list(): every local review manifest. Read-only. load(name): the reconciled view (fresh projection + applied overlay); returns {"error", "type":"NotFound"} for an unknown review. create(name, title?, project?, graph?, question?): create (idempotent) the manifest; initial ordering seeded from the projection order. update(name, title?, question?, project?, graph?, ordering?, set_tier?, exclude?, unexclude?, add_scaffold?, remove_scaffold?, rename_section?, set_section_note?, set_placement?, favorite?, unfavorite?, add_proposed_claim?, remove_proposed_claim?, add_proposed_theme?, remove_proposed_theme?, remove_derived_theme?, restore_derived_theme?): apply any combination of overlay mutations. project/graph change the projection SCOPE (re-projects on next load); set_placement {uid: theme_id} drags a claim into a theme (None returns it to Unplaced); the add_proposed_*/remove_proposed_* families draft themes/claims that live ONLY in the overlay until the user promotes them. rename(name, new_name): change the slug (moves manifest + regenerable sidecars); refuses a colliding name. delete(name): remove the review (durable overlay + regenerable outline). Idempotent; the only NON-regenerable destructive action — gated by the DELETE barrier (refused when ZK_DISABLE_DELETE/WRITE). suggest_claims(name, text, top_k?): rank the corpus's EXISTING claims by similarity to a text topic (read-only). propose_placements(name, granularity?, whole_pool?): propose homes for unplaced claims — an existing theme or a clustered new-theme pitch (read-only). granularity 10..90 (default 50) tunes clustering (low = fewer/broader, high = more/narrower); whole_pool clusters EVERY in-scope claim as fresh topics.

All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def review(
    action: Literal[
        "list", "load", "create", "update", "rename", "delete",
        "suggest_claims", "propose_placements",
    ],
    name: str = "",
    *,
    new_name: str = "",
    title: str = "",
    project: str = "",
    graph: str = "",
    question: str = "",
    ordering: list | None = None,
    set_tier: dict | None = None,
    exclude: list[str] | None = None,
    unexclude: list[str] | None = None,
    add_scaffold: dict | None = None,
    remove_scaffold: list[str] | None = None,
    rename_section: dict | None = None,
    set_section_note: dict | None = None,
    set_placement: dict | None = None,
    favorite: list[str] | None = None,
    unfavorite: list[str] | None = None,
    add_proposed_claim: dict | None = None,
    remove_proposed_claim: list[str] | None = None,
    add_proposed_theme: dict | None = None,
    remove_proposed_theme: list[str] | None = None,
    remove_derived_theme: list[str] | None = None,
    restore_derived_theme: list[str] | None = None,
    text: str = "",
    top_k: int = 0,
    granularity: int = 50,
    whole_pool: bool = False,
) -> str:
    """Use this when reading or authoring a literature-review overlay (the ``_reviews/`` editorial layer).

    A review is a curated, theme-structured narrative over the corpus with two
    separate halves: the PROJECTION (a deterministic, LLM-free view of the corpus
    into themes + a top-level ``unplaced`` worklist, recomputed on every read) and
    the OVERLAY (the author's durable editorial decisions stored as ID-references —
    the SOURCE OF TRUTH; ``load`` reconciles it onto a fresh projection). Scope is
    LOCAL-REPO ONLY; writes are crash-safe (per-review lock, atomic rename,
    debounced ``zettelkasten-mcp`` commit, path-traversal-validated name).

    Actions — name(required, optional?):
        list(): every local review manifest. Read-only.
        load(name): the reconciled view (fresh projection + applied overlay);
            returns {"error", "type":"NotFound"} for an unknown review.
        create(name, title?, project?, graph?, question?): create (idempotent) the
            manifest; initial ``ordering`` seeded from the projection order.
        update(name, title?, question?, project?, graph?, ordering?, set_tier?, exclude?, unexclude?, add_scaffold?, remove_scaffold?, rename_section?, set_section_note?, set_placement?, favorite?, unfavorite?, add_proposed_claim?, remove_proposed_claim?, add_proposed_theme?, remove_proposed_theme?, remove_derived_theme?, restore_derived_theme?):
            apply any combination of overlay mutations. ``project``/``graph`` change
            the projection SCOPE (re-projects on next load); ``set_placement``
            {uid: theme_id} drags a claim into a theme (``None`` returns it to
            Unplaced); the ``add_proposed_*``/``remove_proposed_*`` families draft
            themes/claims that live ONLY in the overlay until the user promotes them.
        rename(name, new_name): change the slug (moves manifest + regenerable
            sidecars); refuses a colliding name.
        delete(name): remove the review (durable overlay + regenerable outline).
            Idempotent; the only NON-regenerable destructive action — gated by the
            DELETE barrier (refused when ZK_DISABLE_DELETE/WRITE).
        suggest_claims(name, text, top_k?): rank the corpus's EXISTING claims by
            similarity to a ``text`` topic (read-only).
        propose_placements(name, granularity?, whole_pool?): propose homes for
            unplaced claims — an existing theme or a clustered new-theme pitch
            (read-only). ``granularity`` 10..90 (default 50) tunes clustering (low =
            fewer/broader, high = more/narrower); ``whole_pool`` clusters EVERY
            in-scope claim as fresh topics.

    All return JSON.
    """
    from zettelkasten import review as review_engine
    from zettelkasten.graph import list_reviews, load_review

    act = (action or "").strip().lower()
    try:
        if act == "list":
            return json.dumps({"reviews": list_reviews()})

        if act == "load":
            if not name:
                raise ValueError("review('load') requires a review name.")
            manifest = load_review(name)
            if not manifest:
                return json.dumps(
                    {"error": f"Review '{name}' not found.", "type": "NotFound"}
                )
            # Non-builder MCP read: keep DERIVED (cluster / cold-start) tiers placed
            # in ``themes`` (their historical centroid behavior) rather than lifting
            # them into ``suggested_themes`` — the opt-in split is a builder/board +
            # gather surface only.
            state = review_engine.reconcile(
                _get_graph, manifest=manifest, split_derived=False
            )
            return json.dumps(state)

        if act == "create":
            blocked = _writes_blocked()
            if blocked:
                return blocked
            if not name:
                raise ValueError("review('create') requires a review name.")
            manifest = review_engine.create_review(
                name,
                title=title or None,
                project=project,
                graph=graph,
                question=question or None,
                get_graph=_get_graph,
            )
            return json.dumps(manifest)

        if act == "update":
            blocked = _writes_blocked()
            if blocked:
                return blocked
            if not name:
                raise ValueError("review('update') requires a review name.")
            manifest = review_engine.update_review(
                name,
                title=title or None,
                question=question or None,
                project=project or None,
                graph=graph or None,
                ordering=ordering,
                set_tier=set_tier,
                exclude=exclude,
                unexclude=unexclude,
                add_scaffold=add_scaffold,
                remove_scaffold=remove_scaffold,
                rename_section=rename_section,
                set_section_note=set_section_note,
                set_placement=set_placement,
                favorite=favorite,
                unfavorite=unfavorite,
                add_proposed_claim=add_proposed_claim,
                remove_proposed_claim=remove_proposed_claim,
                add_proposed_theme=add_proposed_theme,
                remove_proposed_theme=remove_proposed_theme,
                remove_derived_theme=remove_derived_theme,
                restore_derived_theme=restore_derived_theme,
            )
            return json.dumps(manifest)

        if act == "rename":
            # Slug rename: moves the manifest + its regenerable sidecars
            # (``.md`` / ``.tables.json``) and re-stamps the manifest's ``name``.
            # Regenerable file moves are not a hard delete, so this is NOT gated
            # by the delete barrier. A colliding target → ValueError (surfaced as
            # a structured tool error); a missing source → FileNotFoundError.
            blocked = _writes_blocked()
            if blocked:
                return blocked
            if not name:
                raise ValueError("review('rename') requires a review name.")
            if not new_name:
                raise ValueError("review('rename') requires a new_name.")
            manifest = review_engine.rename_review(name, new_name)
            return json.dumps(manifest)

        if act == "delete":
            # Hard-delete of the durable overlay is the one NON-REGENERABLE
            # destructive action this write tool exposes, so it obeys the same
            # delete barrier as ``delete_note`` (refused at runtime because the
            # tool stays registered for its read/create/update actions).
            blocked = _deletes_blocked()
            if blocked:
                return blocked
            if not name:
                raise ValueError("review('delete') requires a review name.")
            result = review_engine.delete_review(name)
            return json.dumps(result)

        if act == "suggest_claims":
            # The "find existing claims for a topic" half of propose-a-theme:
            # ranks the review corpus's EXISTING claims by similarity to ``text``
            # so the agent can draft a section around them (read-only).
            if not name:
                raise ValueError("review('suggest_claims') requires a review name.")
            if not text:
                raise ValueError("review('suggest_claims') requires a topic text.")
            manifest = load_review(name)
            if not manifest:
                return json.dumps({"error": f"Review '{name}' not found.", "type": "NotFound"})
            from zettelkasten import claim_producer
            g = manifest.get("graph", "") or ""
            p = manifest.get("project", "") or ""
            return json.dumps(claim_producer.suggest_claims_for_topic(
                _get_graph,
                text=text,
                project=("" if g else p),
                graph=g,
                top_k=top_k if top_k > 0 else claim_producer.SUGGEST_MAX_CLAIMS,
            ))

        if act == "propose_placements":
            # The SAME engine the dashboard's "Propose placements" button runs:
            # homes each unplaced claim into the nearest existing theme, or
            # clusters the leftovers into new-theme pitches (read-only). The agent
            # then drafts the approved homes via ``update`` (set_placement /
            # add_proposed_theme).
            if not name:
                raise ValueError("review('propose_placements') requires a review name.")
            manifest = load_review(name)
            if not manifest:
                return json.dumps({"error": f"Review '{name}' not found.", "type": "NotFound"})
            from zettelkasten import claim_producer
            # REBUILD mode homes unplaced claims into EXISTING theme centroids, so
            # keep derived tiers placed in ``themes`` (``split_derived=False``)
            # rather than dropping them to ``unplaced`` to be re-clustered anew.
            view = review_engine.reconcile(
                _get_graph, manifest=manifest, split_derived=False
            )
            pp_themes = view.get("themes") or []
            pp_unplaced = view.get("unplaced") or []
            if whole_pool:
                # Cluster the FULL in-scope claim pool (themes' claims ∪ unplaced,
                # deduped) as fresh topics rather than only the leftovers — the
                # MCP equivalent of the wizard's CREATE-mode discovery.
                seen: set[str] = set()
                pool: list[dict] = []
                for c in (*(cl for t in pp_themes for cl in (t.get("claims") or [])), *pp_unplaced):
                    uid = str(c.get("uid") or "")
                    if uid and uid in seen:
                        continue
                    if uid:
                        seen.add(uid)
                    pool.append(c)
                pp_themes, pp_unplaced = [], pool
            return json.dumps(claim_producer.propose_placements(
                _get_graph,
                themes=pp_themes,
                unplaced=pp_unplaced,
                question=manifest.get("question", "") or "",
                scope=view.get("scope") or {},
                cluster_threshold=claim_producer.cluster_threshold_for_granularity(granularity),
            ))

        raise ValueError(
            f"Unknown review action '{action}'. Expected one of: list, load, "
            "create, update, rename, delete, suggest_claims, propose_placements."
        )
    except Exception as e:
        return _json_tool_error(e)

claim ¶

claim(action: Literal['create', 'delete', 'delete_theme', 'materialize', 'materialize_theme', 'propose', 'test', 'discover_contradictions', 'debate_map', 'find_supersessions'], claim_id: str = '', *, title: str = '', body: str | None = None, supports: list | None = None, contradicts: list | None = None, quotes: list | None = None, members: list | None = None, tags: list[str] | None = None, source: dict | None = None, status: str = '', confidence: float | None = None, theme: str = '', text: str = '', project: str = '', graph: str = '', top_k: int = 0, force: bool = False) -> str

Use this when producing grounded claims and landscape themes in the _cross graph (write + read/compute).

The WRITE companion to the read-only Claims engine (build_claims / claim_anatomy): it AUTHORS new claims but never fabricates text/quotes/ stances — the caller supplies the grounded sentence (title + body), member/evidence ids, and verbatim quotes; the tool wires them into the data model. The three FIRST-CLASS-WRITE actions (create, materialize, materialize_theme) mint real _cross entities and obey the PROPOSE-ONLY barrier (refused under ZK_PROPOSE_ONLY/ZK_DISABLE_WRITE with {"type":"ProposeOnly"} — draft via review('update') instead); the read/compute actions stay available. Scope is LOCAL-REPO ONLY; writes are crash-safe.

Actions — name(required, optional?): create(title, claim_id?, body?, supports?, contradicts?, quotes?, tags?, source?, status?, confidence?, project?): create-or-overwrite a grounded claim note (omit claim_id to mint from the title). supports/contradicts are member refs ({id, graph, confidence?} or bare id) authored as stance edges. WRITE. delete(claim_id): SOFT-delete a claim (status=discarded, reversible) and tombstone its stance edges. NOT a hard delete. delete_theme(claim_id): SOFT-delete a landscape concept hub (reverse of materialize_theme); NOT gated by PROPOSE-ONLY (reversible). materialize(claim_id, title, body?, supports?, contradicts?, quotes?, tags?, source?, confidence?, project?): idempotently accept a claim proposal — create plus verbatim quote notes. Requires a stable claim_id. WRITE. materialize_theme(claim_id, title, members?, project?): idempotently materialize a landscape concept hub whose surveys edges bundle members (work-id strings or {id, graph?}). WRITE. propose(theme, project?, graph?, top_k?, force?): MINE grounded claim candidates for a theme (landscape hub id) for the author to accept. READ/COMPUTE only (cached per theme+graph-signature; force bypasses). test(text, project?, graph?, top_k?): TEST a hypothesis against the corpus — gathers evidence, stance-classifies, returns a verdict (supported|contested|refuted|untested-in-corpus) + for/against tally + a promote descriptor. READ/COMPUTE only (quotes retrieved, never generated). discover_contradictions(project?, graph?): DISCOVER candidate contradictions between claims (precision-first). READ/COMPUTE only; returns ranked proposals with gated promote descriptors. debate_map(project?, graph?): aggregate stance edges into a dashboard tension graph (nodes = claims; camps = connected components over contradicts edges). READ only. find_supersessions(project?, graph?): propose belief-revision supersedes edges (B supersedes A when B contradicts A, is stronger + newer, and support overlaps). READ/COMPUTE only.

On the write actions project auto-claims the minted _cross note for that project (explicit membership); on the read/compute actions it scopes the corpus (graph takes priority). top_k=0 means each action's own engine default. All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def claim(
    action: Literal[
        "create", "delete", "delete_theme", "materialize", "materialize_theme",
        "propose", "test", "discover_contradictions", "debate_map", "find_supersessions",
    ],
    claim_id: str = "",
    *,
    title: str = "",
    body: str | None = None,
    supports: list | None = None,
    contradicts: list | None = None,
    quotes: list | None = None,
    members: list | None = None,
    tags: list[str] | None = None,
    source: dict | None = None,
    status: str = "",
    confidence: float | None = None,
    theme: str = "",
    text: str = "",
    project: str = "",
    graph: str = "",
    top_k: int = 0,
    force: bool = False,
) -> str:
    """Use this when producing grounded claims and landscape themes in the ``_cross`` graph (write + read/compute).

    The WRITE companion to the read-only Claims engine (``build_claims`` /
    ``claim_anatomy``): it AUTHORS new claims but never fabricates text/quotes/
    stances — the caller supplies the grounded sentence (``title`` + ``body``),
    member/evidence ids, and verbatim quotes; the tool wires them into the data
    model. The three FIRST-CLASS-WRITE actions (``create``, ``materialize``,
    ``materialize_theme``) mint real ``_cross`` entities and obey the PROPOSE-ONLY
    barrier (refused under ZK_PROPOSE_ONLY/ZK_DISABLE_WRITE with
    ``{"type":"ProposeOnly"}`` — draft via ``review('update')`` instead); the
    read/compute actions stay available. Scope is LOCAL-REPO ONLY; writes are
    crash-safe.

    Actions — name(required, optional?):
        create(title, claim_id?, body?, supports?, contradicts?, quotes?, tags?, source?, status?, confidence?, project?):
            create-or-overwrite a grounded claim note (omit ``claim_id`` to mint
            from the title). ``supports``/``contradicts`` are member refs
            ({id, graph, confidence?} or bare id) authored as stance edges. WRITE.
        delete(claim_id): SOFT-delete a claim (status=discarded, reversible) and
            tombstone its stance edges. NOT a hard delete.
        delete_theme(claim_id): SOFT-delete a landscape ``concept`` hub (reverse of
            materialize_theme); NOT gated by PROPOSE-ONLY (reversible).
        materialize(claim_id, title, body?, supports?, contradicts?, quotes?, tags?, source?, confidence?, project?):
            idempotently accept a claim proposal — ``create`` plus verbatim
            ``quote`` notes. Requires a stable ``claim_id``. WRITE.
        materialize_theme(claim_id, title, members?, project?): idempotently
            materialize a landscape ``concept`` hub whose ``surveys`` edges bundle
            ``members`` (work-id strings or {id, graph?}). WRITE.
        propose(theme, project?, graph?, top_k?, force?): MINE grounded claim
            candidates for a ``theme`` (landscape hub id) for the author to accept.
            READ/COMPUTE only (cached per theme+graph-signature; ``force`` bypasses).
        test(text, project?, graph?, top_k?): TEST a hypothesis against the corpus —
            gathers evidence, stance-classifies, returns a verdict
            (supported|contested|refuted|untested-in-corpus) + for/against tally +
            a ``promote`` descriptor. READ/COMPUTE only (quotes retrieved, never
            generated).
        discover_contradictions(project?, graph?): DISCOVER candidate contradictions
            between claims (precision-first). READ/COMPUTE only; returns ranked
            ``proposals`` with gated ``promote`` descriptors.
        debate_map(project?, graph?): aggregate stance edges into a dashboard
            tension graph (nodes = claims; camps = connected components over
            contradicts edges). READ only.
        find_supersessions(project?, graph?): propose belief-revision ``supersedes``
            edges (B supersedes A when B contradicts A, is stronger + newer, and
            support overlaps). READ/COMPUTE only.

    On the write actions ``project`` auto-claims the minted ``_cross`` note for
    that project (explicit membership); on the read/compute actions it scopes the
    corpus (``graph`` takes priority). ``top_k=0`` means each action's own engine
    default. All return JSON.
    """
    from zettelkasten import claim_producer

    act = (action or "").strip().lower()
    # Pass confidence through untouched so a legitimate 0.0 is preserved (None is
    # the only "unset" sentinel — the producer uses ``is not None`` everywhere).
    conf = confidence
    try:
        # PROPOSE-ONLY barrier: the three first-class-write actions that mint real
        # ``_cross`` graph claims/hubs are refused so a gated agent must draft via
        # the review overlay propose path instead (the user approves the drafts).
        if act in ("create", "materialize", "materialize_theme"):
            blocked = _propose_only_blocked()
            if blocked:
                return blocked

        if act == "create":
            result = claim_producer.create_claim(
                _get_graph,
                claim_id=claim_id,
                title=title,
                body=body,
                supports=supports,
                contradicts=contradicts,
                tags=tags,
                source=source,
                status=status,
                confidence=conf,
            )
            _maybe_register_claim_cross(result, project)
            return json.dumps(result)

        if act == "delete":
            blocked = _writes_blocked()
            if blocked:
                return blocked
            if not claim_id:
                raise ValueError("claim('delete') requires a claim_id.")
            return json.dumps(claim_producer.delete_claim(_get_graph, claim_id=claim_id))

        if act == "delete_theme":
            blocked = _writes_blocked()
            if blocked:
                return blocked
            if not claim_id:
                raise ValueError("claim('delete_theme') requires a claim_id (the hub id).")
            return json.dumps(claim_producer.delete_theme(_get_graph, theme_id=claim_id))

        if act == "materialize":
            if not claim_id:
                raise ValueError("claim('materialize') requires a claim_id.")
            result = claim_producer.materialize_claim(
                _get_graph,
                claim_id=claim_id,
                title=title,
                body=body,
                supports=supports,
                contradicts=contradicts,
                quotes=quotes,
                tags=tags,
                source=source,
                confidence=conf,
            )
            _maybe_register_claim_cross(result, project)
            return json.dumps(result)

        if act == "materialize_theme":
            result = claim_producer.materialize_theme(
                _get_graph,
                theme_id=claim_id,
                title=title,
                members=members,
                body=body,
                tags=tags,
            )
            _maybe_register_claim_cross(result, project)
            return json.dumps(result)

        if act == "propose":
            if not theme:
                raise ValueError("claim('propose') requires a theme (a landscape hub id).")
            # Delegates to the undecorated propose_claims orchestrator (cache →
            # deterministic GATHER → injectable phrasing agent). READ/COMPUTE only.
            return json.dumps(claim_producer.propose_claims(
                _get_graph,
                project=("" if graph else project),
                graph=graph,
                theme=theme,
                top_k=top_k if top_k > 0 else claim_producer.DEFAULT_TOP_K,
                force=force,
            ))

        if act == "test":
            if not text:
                raise ValueError("claim('test') requires a hypothesis text.")
            # Delegates to the undecorated test_claim engine: EPHEMERAL embed →
            # deterministic GATHER → injectable WRITE-FREE stance agent → verdict.
            # READ/COMPUTE only; promotion is a separate (gated) materialize call.
            return json.dumps(claim_producer.test_claim(
                _get_graph,
                text=text,
                project=("" if graph else project),
                graph=graph,
                top_k=top_k if top_k > 0 else claim_producer.TEST_MAX_CANDIDATES,
            ))

        if act == "discover_contradictions":
            # Delegates to the pure, PROPOSE-ONLY read engine in zettelkasten.claims:
            # cluster → deterministic guards → injectable WRITE-FREE stance agent.
            from zettelkasten import claims as _claims

            return json.dumps(_claims.discover_contradictions(
                _get_graph,
                project=("" if graph else project),
                graph=graph,
            ))

        if act == "debate_map":
            from zettelkasten import claims as _claims

            return json.dumps(_claims.debate_map(
                _get_graph,
                project=("" if graph else project),
                graph=graph,
            ))

        if act == "find_supersessions":
            from zettelkasten import claims as _claims

            return json.dumps(_claims.find_supersessions(
                _get_graph,
                project=("" if graph else project),
                graph=graph,
            ))

        raise ValueError(
            f"Unknown claim action '{action}'. Expected one of: "
            "create, delete, delete_theme, materialize, materialize_theme, propose, "
            "test, discover_contradictions, debate_map, find_supersessions."
        )
    except Exception as e:
        return _json_tool_error(e)

create_project ¶

create_project(name: str, description: str, sources: list[str] | None = None, cross: list[str] | None = None) -> str

Create a project manifest in _projects/.

Projects group multiple source graphs for composite analysis.

Parameters:

Name	Type	Description	Default
`name`	`str`	Project name (kebab-case, e.g. 'portfolio-theory')	required
`description`	`str`	What this project studies	required
`sources`	`list[str] \| None`	Optional initial list of source graph names to include	`None`
`cross`	`list[str] \| None`	Optional initial list of `_cross` note ids the project claims	`None`

Source code in zettelkasten/server.py

def create_project(
    name: str,
    description: str,
    sources: list[str] | None = None,
    cross: list[str] | None = None,
) -> str:
    """Create a project manifest in _projects/.

    Projects group multiple source graphs for composite analysis.

    Args:
        name: Project name (kebab-case, e.g. 'portfolio-theory')
        description: What this project studies
        sources: Optional initial list of source graph names to include
        cross: Optional initial list of ``_cross`` note ids the project claims
    """
    try:
        _validate_name(name)
    except ValueError as e:
        return json.dumps({"error": str(e), "type": "ValidationError"})

    projects_dir = GRAPHS_DIR / "_projects"
    projects_dir.mkdir(parents=True, exist_ok=True)

    filepath = projects_dir / f"{name}.yaml"

    import yaml
    from zettelkasten import organizations as _organizations

    # Serialize on the project's OWNER lock so every _projects/<P>.yaml writer
    # shares ONE lock (no lost-update vs a concurrent default_spine/cross write).
    # The exists-check lives INSIDE the lock so two concurrent creates can't both
    # pass it. ``atomic_write_text`` (temp+rename) matches the other writers — no
    # mix of atomic and non-atomic writers on the same manifest.
    with review_write_lock(_organizations._owner_lock_name("project", name), graphs_dir=GRAPHS_DIR):
        if filepath.exists():
            return json.dumps({"error": f"Project '{name}' already exists.", "type": "AlreadyExists"})
        project_data: dict = {
            "name": name,
            "description": description,
            "sources": sources or [],
            "cross": cross or [],
        }
        atomic_write_text(filepath, yaml.dump(project_data, default_flow_style=False))
    logger.info("created project '%s'", name)
    return json.dumps({"name": name, "path": str(filepath), "message": f"Created project '{name}'"})

add_cross_to_project ¶

add_cross_to_project(project: str, cross_id: str) -> str

Claim a _cross note id for a project (append to its cross: list).

Explicit membership: a synthesis note belongs to a project iff its id is in the manifest's cross: list. The same id may be claimed by several projects (the "crosses projects" case) — one physical note, many references, no copies. Idempotent.

Source code in zettelkasten/server.py

def add_cross_to_project(project: str, cross_id: str) -> str:
    """Claim a ``_cross`` note id for a project (append to its ``cross:`` list).

    Explicit membership: a synthesis note belongs to a project iff its id is in
    the manifest's ``cross:`` list. The same id may be claimed by several
    projects (the "crosses projects" case) — one physical note, many references,
    no copies. Idempotent.
    """
    projects_dir = GRAPHS_DIR / "_projects"
    filepath = projects_dir / f"{project}.yaml"
    if not filepath.exists():
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    cid = (cross_id or "").strip()
    if not cid:
        return json.dumps({"error": "cross_id is required.", "type": "ValidationError"})

    import yaml
    from zettelkasten import organizations as _organizations

    # Serialize on the project's OWNER lock — the SAME (non-reentrant) lock
    # set_default_spine_for_project / delete_organization / ensure_migrated take —
    # so a cross-write and a default_spine-write can never interleave a stale
    # read-modify-write that drops the other's field (lost update on the shared
    # _projects/<P>.yaml manifest). Acquired HERE, not at the dispatch, so EVERY
    # caller is covered: the project(action='add_cross') dispatch AND the auto-claim
    # from the note/claim write paths (_register_project_cross / _maybe_register_
    # claim_cross), which never pass through that dispatch. ``atomic_write_text``
    # matches the other manifest writers (no atomic/non-atomic mix).
    with review_write_lock(_organizations._owner_lock_name("project", project), graphs_dir=GRAPHS_DIR):
        project_data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        if not project_data or not isinstance(project_data, dict):
            return json.dumps({"error": f"Project '{project}' has invalid data.", "type": "ValidationError"})

        cross_list = project_cross_ids(project_data)
        if cid in cross_list:
            return json.dumps({"message": f"'{cid}' is already claimed by project '{project}'.", "cross": cross_list})

        cross_list.append(cid)
        project_data["cross"] = cross_list
        atomic_write_text(filepath, yaml.dump(project_data, default_flow_style=False))
    logger.info("added cross note '%s' to project '%s'", cid, project)
    return json.dumps({
        "project": project, "cross_id": cid, "cross": cross_list,
        "message": f"Claimed '{cid}' for project '{project}'",
    })

remove_cross_from_project ¶

remove_cross_from_project(project: str, cross_id: str) -> str

Release a _cross note id from a project's cross: list.

Removes only the reference (the note itself and every other project's claim on it are untouched). Idempotent: removing an unclaimed id is a no-op.

Source code in zettelkasten/server.py

def remove_cross_from_project(project: str, cross_id: str) -> str:
    """Release a ``_cross`` note id from a project's ``cross:`` list.

    Removes only the reference (the note itself and every other project's claim
    on it are untouched). Idempotent: removing an unclaimed id is a no-op.
    """
    projects_dir = GRAPHS_DIR / "_projects"
    filepath = projects_dir / f"{project}.yaml"
    if not filepath.exists():
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    cid = (cross_id or "").strip()
    import yaml
    from zettelkasten import organizations as _organizations

    # Same OWNER-lock + atomic-write discipline as add_cross_to_project: unify this
    # manifest mutator onto the one lock every _projects/<P>.yaml writer shares so a
    # release can never lose-update against a concurrent default_spine/cross write.
    with review_write_lock(_organizations._owner_lock_name("project", project), graphs_dir=GRAPHS_DIR):
        project_data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        if not project_data or not isinstance(project_data, dict):
            return json.dumps({"error": f"Project '{project}' has invalid data.", "type": "ValidationError"})

        cross_list = project_cross_ids(project_data)
        if cid not in cross_list:
            return json.dumps({"message": f"'{cid}' is not claimed by project '{project}'.", "cross": cross_list})

        cross_list = [c for c in cross_list if c != cid]
        project_data["cross"] = cross_list
        atomic_write_text(filepath, yaml.dump(project_data, default_flow_style=False))
    logger.info("removed cross note '%s' from project '%s'", cid, project)
    return json.dumps({
        "project": project, "cross_id": cid, "cross": cross_list,
        "message": f"Released '{cid}' from project '{project}'",
    })

set_default_spine_for_project ¶

set_default_spine_for_project(project: str, org_id: str) -> str

Set (or clear) a project's default_spine pointer (the primary org).

The spine directory's counterpart to :func:add_cross_to_project: org_id names the project-owned org that is the project's primary organizing principle (its "Organize by" default). An EMPTY org_id clears the pointer back to the implicit intrinsic lens (the organic cluster graph, which has no org record). Idempotent. Mirrors the cross write path: load → set → rewrite the manifest, under the project's write lock.

A non-empty org_id is VALIDATED before it is written: it must resolve to an existing project-owned org, so the pointer can never be set to a name that dangles from the moment it is written (the deletion-time clears in organizations guard the other direction). An empty org_id always clears.

LOCKING (defects A+B). The whole validate-then-write runs under the project's OWNER lock — review_write_lock(organizations._owner_lock_name('project', P)) — the EXACT same (non-reentrant) lock delete_organization / save_organization / ensure_migrated take. This guarantees set and the clear-on-delete are MUTUALLY EXCLUSIVE (a delete that clears default_spine can never interleave between this function's existence check and its pointer write), and that exactly ONE lock is held (the dispatch no longer wraps this in project::P, so there is no double-acquire). Because the owner lock is taken here, the existence re-check inside it uses migrate=False (a migrate=True load would re-enter ensure_migrated and re-acquire this very lock → deadlock); any un-absorbed legacy tables are migrated ONCE up front, outside the lock.

Source code in zettelkasten/server.py

def set_default_spine_for_project(project: str, org_id: str) -> str:
    """Set (or clear) a project's ``default_spine`` pointer (the primary org).

    The spine directory's counterpart to :func:`add_cross_to_project`: ``org_id``
    names the project-owned org that is the project's primary organizing principle
    (its "Organize by" default). An EMPTY ``org_id`` clears the pointer back to the
    implicit intrinsic lens (the organic cluster graph, which has no org record).
    Idempotent. Mirrors the cross write path: load → set → rewrite the manifest,
    under the project's write lock.

    A non-empty ``org_id`` is VALIDATED before it is written: it must resolve to an
    existing project-owned org, so the pointer can never be set to a name that
    dangles from the moment it is written (the deletion-time clears in
    ``organizations`` guard the other direction). An empty ``org_id`` always clears.

    LOCKING (defects A+B). The whole validate-then-write runs under the project's
    OWNER lock — ``review_write_lock(organizations._owner_lock_name('project', P))``
    — the EXACT same (non-reentrant) lock ``delete_organization`` /
    ``save_organization`` / ``ensure_migrated`` take. This guarantees set and the
    clear-on-delete are MUTUALLY EXCLUSIVE (a delete that clears ``default_spine``
    can never interleave between this function's existence check and its pointer
    write), and that exactly ONE lock is held (the dispatch no longer wraps this in
    project::P, so there is no double-acquire). Because the owner lock is taken
    here, the existence re-check inside it uses ``migrate=False`` (a ``migrate=True``
    load would re-enter ``ensure_migrated`` and re-acquire this very lock → deadlock);
    any un-absorbed legacy tables are migrated ONCE up front, outside the lock.
    """
    projects_dir = GRAPHS_DIR / "_projects"
    filepath = projects_dir / f"{project}.yaml"
    if not filepath.exists():
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    import yaml
    from zettelkasten import organizations as _organizations

    oid = (org_id or "").strip()
    # Pre-migrate OUTSIDE the owner lock so the existence re-check below can run
    # lock-free (migrate=False). Doing a migrate=True load under the owner lock
    # would re-acquire it via ``ensure_migrated`` and deadlock. No-op when there
    # are no un-absorbed legacy tables; best-effort (a migration failure must not
    # block setting a pointer at an already-materialized org).
    if oid:
        try:
            _organizations.ensure_migrated("project", project, graphs_dir=GRAPHS_DIR)
        except Exception:  # noqa: BLE001 — best-effort; validation below still guards
            logger.warning(
                "ensure_migrated failed before set_default_spine for project '%s'",
                project, exc_info=True,
            )

    owner_lock = _organizations._owner_lock_name("project", project)
    with review_write_lock(owner_lock, graphs_dir=GRAPHS_DIR):
        project_data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        if not project_data or not isinstance(project_data, dict):
            return json.dumps({"error": f"Project '{project}' has invalid data.", "type": "ValidationError"})

        if oid:
            # Re-validate existence IMMEDIATELY before the write, under the lock, so
            # a delete that ran first is observed and we refuse rather than write a
            # dangling pointer. ``migrate=False`` — the pre-migrate above absorbed
            # any legacy tables; a migrate=True load here would re-acquire this lock.
            if _organizations.load_organization(
                "project", project, oid, graphs_dir=GRAPHS_DIR, migrate=False
            ) is None:
                return json.dumps({
                    "error": f"Organization '{oid}' not found for project '{project}'.",
                    "type": "NotFound",
                })
            project_data["default_spine"] = oid
        else:
            # Empty pointer == intrinsic lens: drop the field rather than store "".
            project_data.pop("default_spine", None)
        # Atomic (temp+rename) write, matching every other _projects/<P>.yaml writer
        # — no mix of atomic and non-atomic writers on the shared manifest.
        atomic_write_text(filepath, yaml.dump(project_data, default_flow_style=False))
        current = project_default_spine(project_data)
    if oid:
        logger.info("set default_spine '%s' for project '%s'", oid, project)
        msg = f"Set default spine '{oid}' for project '{project}'"
    else:
        logger.info("cleared default_spine for project '%s'", project)
        msg = f"Cleared default spine for project '{project}' (intrinsic lens)"
    return json.dumps({
        "project": project, "default_spine": current, "message": msg,
    })

add_to_project ¶

add_to_project(project: str, source: str) -> str

Add a source graph to a project.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name	required
`source`	`str`	Source graph name to add	required

Source code in zettelkasten/server.py

def add_to_project(project: str, source: str) -> str:
    """Add a source graph to a project.

    Args:
        project: Project name
        source: Source graph name to add
    """
    projects_dir = GRAPHS_DIR / "_projects"
    filepath = projects_dir / f"{project}.yaml"
    if not filepath.exists():
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    import yaml
    from zettelkasten import organizations as _organizations

    # Serialize the manifest read-modify-write on the project's OWNER lock so a
    # source-add can't lose-update a concurrent default_spine/cross write that
    # touches the SAME _projects/<P>.yaml (``add_to_project`` round-trips the whole
    # dict, so a stale read would drop the other writer's field). The lock scope is
    # JUST the RMW; the read-only synthesis follow-up below runs unlocked (it never
    # writes the manifest, and holding the owner lock through it is needless).
    with review_write_lock(_organizations._owner_lock_name("project", project), graphs_dir=GRAPHS_DIR):
        project_data = yaml.safe_load(filepath.read_text(encoding="utf-8"))
        if not project_data or not isinstance(project_data, dict):
            return json.dumps({"error": f"Project '{project}' has invalid data.", "type": "ValidationError"})

        sources_list = project_data.get("sources", [])
        if source in sources_list:
            return json.dumps({"message": f"Source '{source}' is already in project '{project}'.", "sources": sources_list})

        sources_list.append(source)
        project_data["sources"] = sources_list
        atomic_write_text(filepath, yaml.dump(project_data, default_flow_style=False))

    logger.info("added source '%s' to project '%s'", source, project)
    response: dict = {"project": project, "source": source, "sources": sources_list, "message": f"Added '{source}' to project '{project}'"}

    # Auto-suggest synthesis follow-up: now that a source joined the project, see
    # whether its notes feed existing _cross hubs (attach) or form new cross-source
    # clusters (new_hub). This is the cheap incremental check (only the new source
    # is examined), so it is safe to run on every add. It only *suggests* — writing
    # synthesis notes stays an agent judgment call. Never let it fail the add.
    try:
        followup = json.loads(suggest_concept_hubs(project, since_source=source))
        if followup.get("suggestions"):
            response["synthesis_followup"] = {
                "counts": followup.get("counts", {}),
                "suggestions": followup["suggestions"],
                "guidance": (
                    "Act on these to keep synthesis current: prefer action='attach' "
                    "(link the note into the existing _cross hub) over action='new_hub'. "
                    "Only mint a new _cross note for clusters with no existing hub."
                ),
            }
    except Exception as exc:  # noqa: BLE001 - advisory only, must not block the add
        logger.warning("synthesis follow-up skipped for '%s'/'%s': %s", project, source, exc)

    return json.dumps(response)

get_extraction_protocol ¶

get_extraction_protocol(doc_type: str) -> dict

Get extraction guidance for a document type.

Returns recommended note types to extract and strategies for different kinds of source material, plus a cross-cutting evidence protocol describing how notes must be backed by direct quotes (preferring the user's Zotero highlights) and that quotes are RETRIEVED, never generated. create_source embeds this automatically; call this directly to re-fetch the protocol mid-extraction without recreating the source.

Parameters:

Name	Type	Description	Default
`doc_type`	`str`	One of 'journal-article', 'book', 'working-paper', 'report', 'review', 'blog'	required

Source code in zettelkasten/server.py

def get_extraction_protocol(doc_type: str) -> dict:
    """Get extraction guidance for a document type.

    Returns recommended note types to extract and strategies for different kinds
    of source material, plus a cross-cutting evidence protocol describing how
    notes must be backed by direct quotes (preferring the user's Zotero
    highlights) and that quotes are RETRIEVED, never generated. create_source
    embeds this automatically; call this directly to re-fetch the protocol
    mid-extraction without recreating the source.

    Args:
        doc_type: One of 'journal-article', 'book', 'working-paper', 'report',
            'review', 'blog'
    """
    return _extraction_protocol(doc_type)

zotero_lookup ¶

zotero_lookup(query: str) -> str

Search the user's Zotero library by title or author.

Requires ZOTERO_USER_ID and ZOTERO_API_KEY env vars to be configured.

Parameters:

Name	Type	Description	Default
`query`	`str`	Search term (title, author, keyword)	required

Source code in zettelkasten/server.py

def zotero_lookup(query: str) -> str:
    """Search the user's Zotero library by title or author.

    Requires ZOTERO_USER_ID and ZOTERO_API_KEY env vars to be configured.

    Args:
        query: Search term (title, author, keyword)
    """
    from zettelkasten.zotero import search
    results = search(query)
    return json.dumps({"query": query, "results": results, "count": len(results)})

zotero_fetch_pdf ¶

zotero_fetch_pdf(key: str, include_text: bool = True, include_annotations: bool = True) -> str

Fetch a paper's full-text PDF from the LOCAL Zotero install.

Resolves a Zotero item key (the zotero_key stored on a source or citation) to its PDF attachment on disk, returning the absolute file path plus the extracted text and any saved highlights/notes. Reads the local Zotero data dir (~/Zotero, override with ZOTERO_DATA_DIR) directly — no API key needed and works offline. Unlike zotero_lookup (Web API, metadata only), this gives full text for deep extraction into the zettelkasten.

Parameters:

Name	Type	Description	Default
`key`	`str`	Zotero item key (top-level item; its PDF attachment is resolved automatically)	required
`include_text`	`bool`	Extract the PDF's text (needs pypdf; falls back to path-only with a warning)	`True`
`include_annotations`	`bool`	Include saved highlights/notes from local storage	`True`

Source code in zettelkasten/server.py

def zotero_fetch_pdf(key: str, include_text: bool = True, include_annotations: bool = True) -> str:
    """Fetch a paper's full-text PDF from the LOCAL Zotero install.

    Resolves a Zotero item key (the `zotero_key` stored on a source or citation)
    to its PDF attachment on disk, returning the absolute file path plus the
    extracted text and any saved highlights/notes. Reads the local Zotero data
    dir (~/Zotero, override with ZOTERO_DATA_DIR) directly — no API key needed
    and works offline. Unlike `zotero_lookup` (Web API, metadata only), this
    gives full text for deep extraction into the zettelkasten.

    Args:
        key: Zotero item key (top-level item; its PDF attachment is resolved automatically)
        include_text: Extract the PDF's text (needs pypdf; falls back to path-only with a warning)
        include_annotations: Include saved highlights/notes from local storage
    """
    from zettelkasten.zotero import fetch_pdf
    result = fetch_pdf(key, include_text=include_text, include_annotations=include_annotations)
    return json.dumps(result)

zotero_collections ¶

zotero_collections() -> str

List the Zotero library's collections (folders) for browsing.

Local-first (reads ~/Zotero, override with ZOTERO_DATA_DIR) with a Web API fallback when ZOTERO_USER_ID/ZOTERO_API_KEY are set. Each collection reports its key, name, full path, parent key, and top-level item count — so you can see how the library is organized instead of only guessing keyword searches.

Source code in zettelkasten/server.py

def zotero_collections() -> str:
    """List the Zotero library's collections (folders) for browsing.

    Local-first (reads ~/Zotero, override with ZOTERO_DATA_DIR) with a Web API
    fallback when ZOTERO_USER_ID/ZOTERO_API_KEY are set. Each collection reports
    its key, name, full path, parent key, and top-level item count — so you can
    see how the library is organized instead of only guessing keyword searches.
    """
    from zettelkasten.zotero import list_collections
    return json.dumps(list_collections())

zotero_list_items ¶

zotero_list_items(collection: str = '', limit: int = 200) -> str

List the papers in a Zotero collection (or the whole library).

Parameters:

Name	Type	Description	Default
`collection`	`str`	A collection key (preferred) or name (case-insensitive). Empty lists the most recently added top-level items library-wide.	`''`
`limit`	`int`	Maximum items to return (default 200).	`200`

Source code in zettelkasten/server.py

def zotero_list_items(collection: str = "", limit: int = 200) -> str:
    """List the papers in a Zotero collection (or the whole library).

    Args:
        collection: A collection key (preferred) or name (case-insensitive).
            Empty lists the most recently added top-level items library-wide.
        limit: Maximum items to return (default 200).
    """
    from zettelkasten.zotero import list_items
    return json.dumps(list_items(collection=collection, limit=limit))

backfill_quotes ¶

backfill_quotes(source: str = '', apply: bool = False, link_floor: float = 0.45) -> str

Retroactively capture verbatim quote notes from saved Zotero highlights.

Sources extracted before the quote-evidence protocol carry paraphrased claim/finding notes but no verbatim quotes. This walks such sources, pulls the user's saved Zotero highlights, and materializes each as a quote note whose body is the verbatim text and whose source.page is the highlight's pageLabel — linked supports to the claim/finding it best matches (by embedding similarity, the same backend as semantic_search). Highlights are RETRIEVED, never generated; a highlight below link_floor is still preserved as an unlinked quote rather than mis-attributed. Idempotent: a highlight already captured as a quote note is skipped.

Parameters:

Name	Type	Description	Default
`source`	`str`	A single source folder to backfill; empty = every source with a zotero_key.	`''`
`apply`	`bool`	Dry run when False (default) — reports the plan without writing. Set True to actually create the quote notes and their links.	`False`
`link_floor`	`float`	Minimum similarity (0-1) to link a highlight to a claim; below it the quote is created unlinked.	`0.45`

Source code in zettelkasten/server.py

def backfill_quotes(source: str = "", apply: bool = False, link_floor: float = 0.45) -> str:
    """Retroactively capture verbatim quote notes from saved Zotero highlights.

    Sources extracted before the quote-evidence protocol carry paraphrased
    claim/finding notes but no verbatim quotes. This walks such sources, pulls the
    user's saved Zotero highlights, and materializes each as a `quote` note whose
    body is the verbatim text and whose source.page is the highlight's pageLabel —
    linked `supports` to the claim/finding it best matches (by embedding
    similarity, the same backend as semantic_search). Highlights are RETRIEVED,
    never generated; a highlight below `link_floor` is still preserved as an
    unlinked quote rather than mis-attributed. Idempotent: a highlight already
    captured as a quote note is skipped.

    Args:
        source: A single source folder to backfill; empty = every source with a
            zotero_key.
        apply: Dry run when False (default) — reports the plan without writing.
            Set True to actually create the quote notes and their links.
        link_floor: Minimum similarity (0-1) to link a highlight to a claim;
            below it the quote is created unlinked.
    """
    from zettelkasten.backfill import backfill_quotes as _backfill

    result = _backfill(_get_graph, source=source, apply=apply, link_floor=link_floor)
    return json.dumps(result)

ingest_source ¶

ingest_source(key: str = '', path: str = '', title: str = '', max_chars: int = 200000, cache_full: bool = False) -> str

Canonical source front door: resolve a paper to full text + a stable hash.

Provide exactly one of: key (a Zotero item key), path (a local file — absolute, cwd-relative, or a bare filename under ZK_SOURCES_DIR), or title (looked up in Zotero; a single match resolves automatically, otherwise the candidate list is returned). Extracts the full text and any saved annotations, computes a stable content_hash (sha256 of the file bytes), caches the text for later offline reuse, and reports existing_source when a source already carries that hash.

Feed content_hash, source_path, and zotero_key from the result into create_source.

Parameters:

Name	Type	Description	Default
`key`	`str`	Zotero item key (preferred path; find it via zotero_lookup)	`''`
`path`	`str`	Local file path (fallback path)	`''`
`title`	`str`	Title to look up in Zotero	`''`
`max_chars`	`int`	Maximum characters of text to extract (default 200000)	`200000`
`cache_full`	`bool`	Cache the WHOLE document (plus a page/outline sidecar) while still returning at most `max_chars`. Required for book-scale sources so a later chapter is addressable via a windowed `fulltext` read.	`False`

Source code in zettelkasten/server.py

def ingest_source(
    key: str = "",
    path: str = "",
    title: str = "",
    max_chars: int = 200_000,
    cache_full: bool = False,
) -> str:
    """Canonical source front door: resolve a paper to full text + a stable hash.

    Provide exactly one of: `key` (a Zotero item key), `path` (a local file —
    absolute, cwd-relative, or a bare filename under ZK_SOURCES_DIR), or `title`
    (looked up in Zotero; a single match resolves automatically, otherwise the
    candidate list is returned). Extracts the full text and any saved annotations,
    computes a stable `content_hash` (sha256 of the file bytes), caches the text
    for later offline reuse, and reports `existing_source` when a source already
    carries that hash.

    Feed `content_hash`, `source_path`, and `zotero_key` from the result into
    `create_source`.

    Args:
        key: Zotero item key (preferred path; find it via zotero_lookup)
        path: Local file path (fallback path)
        title: Title to look up in Zotero
        max_chars: Maximum characters of text to extract (default 200000)
        cache_full: Cache the WHOLE document (plus a page/outline sidecar) while
            still returning at most `max_chars`. Required for book-scale sources
            so a later chapter is addressable via a windowed `fulltext` read.
    """
    from zettelkasten.ingest import ingest

    result = ingest(
        key=key, path=path, title=title, max_chars=max_chars, cache_full=cache_full
    )
    if "error" not in result:
        existing = _find_source_by_hash(result.get("content_hash", ""))
        if existing:
            result["existing_source"] = existing
    return json.dumps(result)

get_fulltext ¶

get_fulltext(source: str, max_chars: int = 200000, offset: int = 0, page_start: 'int | None' = None, page_end: 'int | None' = None) -> str

Return a source's cached full text (or a window of it), re-deriving cold.

Reads the disposable cache at .angelo/zettel/fulltext/.txt. If the cache was cleared, re-derives the text from the source's committed pointer (source_path, else zotero_key) and repopulates the cache — so the reference is reproducible offline without re-fetching from Zotero.

For book-scale sources the cache holds the WHOLE document; pass offset (char start) or a page_start/page_end range (0-based, inclusive end) to read a later slice. The result reports total_chars, num_pages, and sliced so a reader knows there is more beyond the returned window.

Parameters:

Name	Type	Description	Default
`source`	`str`	Source graph name (folder under .zettelkasten/)	required
`max_chars`	`int`	Maximum characters to return in this window (default 200000)	`200000`
`offset`	`int`	Char index to start reading from (default 0). Ignored when a page range is given and a page table exists.	`0`
`page_start`	`'int \| None'`	0-based first page of the window (needs a cached page index).	`None`
`page_end`	`'int \| None'`	0-based last page of the window, inclusive.	`None`

Source code in zettelkasten/server.py

def get_fulltext(
    source: str,
    max_chars: int = 200_000,
    offset: int = 0,
    page_start: "int | None" = None,
    page_end: "int | None" = None,
) -> str:
    """Return a source's cached full text (or a window of it), re-deriving cold.

    Reads the disposable cache at .angelo/zettel/fulltext/<content_hash>.txt. If
    the cache was cleared, re-derives the text from the source's committed pointer
    (`source_path`, else `zotero_key`) and repopulates the cache — so the
    reference is reproducible offline without re-fetching from Zotero.

    For book-scale sources the cache holds the WHOLE document; pass `offset` (char
    start) or a `page_start`/`page_end` range (0-based, inclusive end) to read a
    later slice. The result reports `total_chars`, `num_pages`, and `sliced` so a
    reader knows there is more beyond the returned window.

    Args:
        source: Source graph name (folder under .zettelkasten/)
        max_chars: Maximum characters to return in this window (default 200000)
        offset: Char index to start reading from (default 0). Ignored when a page
            range is given and a page table exists.
        page_start: 0-based first page of the window (needs a cached page index).
        page_end: 0-based last page of the window, inclusive.
    """
    from zettelkasten.ingest import ingest, read_cached_fulltext, read_page_index

    meta = load_source_meta(source)
    if not meta:
        return json.dumps({"error": f"Source '{source}' not found.", "type": "NotFound"})

    content_hash = meta.get("content_hash", "")
    if content_hash:
        cached = read_cached_fulltext(content_hash)
        if cached is not None:
            idx = read_page_index(content_hash) or {}
            page_offsets = idx.get("page_offsets") or None
            total = len(cached)
            start, end = _resolve_char_window(
                total, max_chars, offset, page_start, page_end, page_offsets
            )
            text = cached[start:end]
            return json.dumps({
                "source": source,
                "content_hash": content_hash,
                "text": text,
                "cached": True,
                "total_chars": total,
                "num_pages": idx.get("num_pages", 0),
                "offset": start,
                "end": end,
                "sliced": start > 0 or end < total,
            })

    source_path = meta.get("source_path", "")
    zotero_key = meta.get("zotero_key", "")
    if not source_path and not zotero_key:
        return json.dumps({
            "error": f"Source '{source}' has no cached text and no pointer "
            "(content_hash/source_path/zotero_key) to re-derive it from. "
            "Re-ingest it with ingest_source.",
            "type": "NoReference",
        })

    # Cold cache: re-derive the WHOLE document (cache_full) so the sidecar/page
    # table is rebuilt and a windowed read is possible, then slice it here.
    windowed = offset or page_start is not None or page_end is not None
    derived = ingest(
        key=zotero_key if not source_path else "",
        path=source_path,
        max_chars=max_chars,
        cache_full=bool(windowed),
    )
    if "error" in derived:
        return json.dumps(derived)
    derived_hash = derived.get("content_hash", content_hash)
    text = derived.get("text", "")
    if windowed and derived_hash:
        cached = read_cached_fulltext(derived_hash)
        if cached is not None:
            idx = read_page_index(derived_hash) or {}
            page_offsets = idx.get("page_offsets") or None
            total = len(cached)
            start, end = _resolve_char_window(
                total, max_chars, offset, page_start, page_end, page_offsets
            )
            return json.dumps({
                "source": source,
                "content_hash": derived_hash,
                "text": cached[start:end],
                "cached": False,
                "re_derived_from": source_path or f"zotero:{zotero_key}",
                "total_chars": total,
                "num_pages": idx.get("num_pages", 0),
                "offset": start,
                "end": end,
                "sliced": start > 0 or end < total,
            })
    return json.dumps({
        "source": source,
        "content_hash": derived_hash,
        "text": text,
        "cached": False,
        "re_derived_from": source_path or f"zotero:{zotero_key}",
        "total_chars": derived.get("total_chars", len(text)),
    })

get_source_outline ¶

get_source_outline(source: str) -> str

Return a source's cached chapter outline + page index, if any.

Reads the disposable <content_hash>.pages.json sidecar. Returns {source, content_hash, num_pages, total_chars, outline} where each outline entry is {title, page, char_offset}. outline is empty for sources with no detectable bookmarks (papers, un-bookmarked PDFs, plain text) — the caller then falls back to char-window slicing.

Source code in zettelkasten/server.py

def get_source_outline(source: str) -> str:
    """Return a source's cached chapter outline + page index, if any.

    Reads the disposable `<content_hash>.pages.json` sidecar. Returns
    `{source, content_hash, num_pages, total_chars, outline}` where each outline
    entry is `{title, page, char_offset}`. `outline` is empty for sources with no
    detectable bookmarks (papers, un-bookmarked PDFs, plain text) — the caller
    then falls back to char-window slicing.
    """
    from zettelkasten.ingest import read_page_index

    meta = load_source_meta(source)
    if not meta:
        return json.dumps({"error": f"Source '{source}' not found.", "type": "NotFound"})
    content_hash = meta.get("content_hash", "")
    idx = read_page_index(content_hash) if content_hash else None
    if not idx:
        return json.dumps({
            "source": source,
            "content_hash": content_hash,
            "num_pages": 0,
            "total_chars": 0,
            "outline": [],
        })
    return json.dumps({
        "source": source,
        "content_hash": content_hash,
        "num_pages": idx.get("num_pages", 0),
        "total_chars": idx.get("total_chars", 0),
        "outline": idx.get("outline", []),
    })

project_search ¶

project_search(query: str, top_k: int = 10, project: str = '', since: str = '', until: str = '') -> str

Semantic search across all sources in a project (merged embeddings).

If project is empty, searches ALL source graphs + _cross/ globally. When the global ANN index is enabled (default), a broad scope is served by a single pre-filtered vector search rather than a per-box fan-out.

Parameters:

Name	Type	Description	Default
`query`	`str`	Natural language search query	required
`top_k`	`int`	Number of results to return (default 10)	`10`
`project`	`str`	Project name (empty = search all graphs)	`''`
`since`	`str`	Optional inclusive lower date bound (`2026`, `2026-03`, or `2026-03-15`) filtering on each source box's date.	`''`
`until`	`str`	Optional inclusive upper date bound (filled to the end of its period, so `until='2026'` includes all of 2026).	`''`

Source code in zettelkasten/server.py

def project_search(
    query: str, top_k: int = 10, project: str = "", since: str = "", until: str = ""
) -> str:
    """Semantic search across all sources in a project (merged embeddings).

    If project is empty, searches ALL source graphs + _cross/ globally. When the
    global ANN index is enabled (default), a broad scope is served by a single
    pre-filtered vector search rather than a per-box fan-out.

    Args:
        query: Natural language search query
        top_k: Number of results to return (default 10)
        project: Project name (empty = search all graphs)
        since: Optional inclusive lower date bound (``2026``, ``2026-03``, or
            ``2026-03-15``) filtering on each source box's date.
        until: Optional inclusive upper date bound (filled to the end of its
            period, so ``until='2026'`` includes all of 2026).
    """
    if not query.strip():
        return json.dumps({"error": "query must not be empty.", "type": "ValidationError"})

    source_names: list[str] = []
    if project:
        project_data = load_project(project)
        if not project_data:
            return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})
        source_names = project_data.get("sources", [])
        cross_dir = GRAPHS_DIR / "_cross"
        if cross_dir.is_dir():
            source_names = source_names + ["_cross"]
    else:
        if GRAPHS_DIR.exists():
            for d in sorted(GRAPHS_DIR.iterdir()):
                if not d.is_dir():
                    continue
                if d.name.startswith("_") and d.name != "_cross":
                    continue
                source_names.append(d.name)

    from zettelkasten import config as zk_config

    cfg = zk_config.rerank_config()
    disabled = zk_config.rerank_disabled(cfg)
    recall_k = top_k if disabled else max(top_k, cfg.overfetch * top_k)
    date_range = _parse_date_range(since, until)
    all_results = _semantic_recall(
        query, source_names, recall_k, date_range=date_range, whole_corpus=not project
    )

    if disabled:
        all_results.sort(key=lambda x: x[1], reverse=True)
        top_results = all_results[:top_k]
    else:
        top_results = _rerank_semantic_hits(all_results, top_k=top_k, cfg=cfg)

    results = []
    for note_id, score, source_name in top_results:
        zg = _get_graph(source_name)
        note = zg.notes[note_id]
        results.append({
            "id": note_id,
            "title": note.title,
            "type": note.type,
            "source_graph": source_name,
            "similarity": round(score, 3),
            "body_preview": note.body[:150] + "..." if len(note.body) > 150 else note.body,
            "tags": note.tags,
        })

    return json.dumps({"query": query, "project": project or "(all)", "results": results, "count": len(results)})

query_topic ¶

query_topic(project: str, topic: str, top_k: int = 15, raw: bool = False, since: str = '', until: str = '') -> str

Use this when you want to research a topic across project sources and get structured output.

Finds relevant notes, groups by type, identifies disagreements between sources, and flags coverage warnings for under-reviewed sources.

Parameters:

Name	Type	Description	Default
`project`	`str`	Project name (with raw=True, empty = search ALL graphs)	required
`topic`	`str`	Research topic or question	required
`top_k`	`int`	Number of notes to retrieve (default 15)	`15`
`raw`	`bool`	If true, return a flat semantic-search result list across the project's sources (no grouping/disagreement analysis)	`False`
`since`	`str`	Optional inclusive lower date bound (`2026`, `2026-03`, or `2026-03-15`) filtering on each source box's date.	`''`
`until`	`str`	Optional inclusive upper date bound (filled to the end of its period, so `until='2026'` includes all of 2026).	`''`

Source code in zettelkasten/server.py

@mcp_server.tool()
def query_topic(
    project: str, topic: str, top_k: int = 15, raw: bool = False,
    since: str = "", until: str = "",
) -> str:
    """Use this when you want to research a topic across project sources and get structured output.

    Finds relevant notes, groups by type, identifies disagreements between
    sources, and flags coverage warnings for under-reviewed sources.

    Args:
        project: Project name (with raw=True, empty = search ALL graphs)
        topic: Research topic or question
        top_k: Number of notes to retrieve (default 15)
        raw: If true, return a flat semantic-search result list across the
            project's sources (no grouping/disagreement analysis)
        since: Optional inclusive lower date bound (``2026``, ``2026-03``, or
            ``2026-03-15``) filtering on each source box's date.
        until: Optional inclusive upper date bound (filled to the end of its
            period, so ``until='2026'`` includes all of 2026).
    """
    import yaml as _yaml

    if not topic.strip():
        return json.dumps({"error": "topic must not be empty.", "type": "ValidationError"})

    if raw:
        return project_search(query=topic, top_k=top_k, project=project, since=since, until=until)

    project_data = load_project(project)
    if not project_data:
        return json.dumps({"error": f"Project '{project}' not found.", "type": "NotFound"})

    source_names = project_data.get("sources", [])
    cross_dir = GRAPHS_DIR / "_cross"
    search_sources = list(source_names)
    if cross_dir.is_dir():
        search_sources.append("_cross")

    from zettelkasten import config as zk_config

    cfg = zk_config.rerank_config()
    disabled = zk_config.rerank_disabled(cfg)
    recall_k = top_k if disabled else max(top_k, cfg.overfetch * top_k)
    date_range = _parse_date_range(since, until)

    all_hits = _semantic_recall(topic, search_sources, recall_k, date_range=date_range)

    if disabled:
        all_hits.sort(key=lambda x: x[1], reverse=True)
        top_hits = all_hits[:top_k]
    else:
        # Reserve per-type slots so hub-importance boosting never squeezes a whole
        # type (e.g. findings) out of the global top_k before the grouping below.
        top_hits = _rerank_semantic_hits(
            all_hits, top_k=top_k, cfg=cfg, reserve_by_type=True
        )

    grouped: dict[str, list[dict]] = {}
    # Key the result set on the COMPOSITE (source_graph, note_id): zettel ids are
    # not namespaced by box, so a bare-id set would phantom-match a same-id note
    # from a different source graph and flip a disagreement on/off. A note's links
    # target ids within that note's own source_graph, so compare
    # (source_name, link.target) against this composite set.
    result_ids = {(source_name, note_id) for note_id, _, source_name in top_hits}
    disagreements: list[dict] = []

    for note_id, score, source_name in top_hits:
        zg = _get_graph(source_name)
        note = zg.notes[note_id]
        entry = {
            "id": note_id,
            "title": note.title,
            "source_graph": source_name,
            "similarity": round(score, 3),
            "body_preview": note.body[:200] + "..." if len(note.body) > 200 else note.body,
        }
        grouped.setdefault(note.type, []).append(entry)

        for link in note.links:
            # The edge's target lives in ``link.graph`` (empty => the linker's own
            # ``source_name``), NOT necessarily in ``source_name`` — so resolve the
            # target-side composite key from the per-edge graph. This preserves the
            # phantom guard (empty ``link.graph`` yields the same ``(source_name,
            # target)`` key as before) while letting a genuine CROSS-GRAPH stance
            # edge match a member living in another source graph.
            if link.relation in ("contradicts", "qualifies") and (link.graph or source_name, link.target) in result_ids:
                disagreements.append({
                    "from_note": note_id,
                    "from_title": note.title,
                    "relation": link.relation,
                    "to_note": link.target,
                    "source_graph": source_name,
                })

    coverage_warnings: list[dict] = []
    for source_name in source_names:
        meta_path = GRAPHS_DIR / source_name / "_meta.yaml"
        if not meta_path.exists():
            continue
        meta = _yaml.safe_load(meta_path.read_text(encoding="utf-8")) or {}
        coverage = meta.get("coverage", {})
        human_cov = coverage.get("human", "none")
        agent_cov = coverage.get("agent", "none")
        if human_cov == "none" and agent_cov != "none":
            warning = {
                "source": source_name,
                "warning": "agent-only extraction (no human review)",
                "coverage": coverage,
            }
            audit = _verify_source_extraction(source_name)
            if "error" not in audit and audit.get("blocking_count", 0) > 0:
                warning["grounding_violations"] = audit["blocking_count"]
            coverage_warnings.append(warning)

    return json.dumps({
        "topic": topic,
        "project": project,
        "findings": grouped.get("finding", []),
        "claims": grouped.get("claim", []),
        "methods": grouped.get("method", []),
        "concepts": grouped.get("concept", []),
        "other": {k: v for k, v in grouped.items() if k not in ("finding", "claim", "method", "concept")},
        "disagreements": disagreements,
        "coverage_warnings": coverage_warnings,
        "total_results": len(top_hits),
    })

frame_question ¶

frame_question(question: str, project: str = '', facets: list[str] | None = None, max_facets: int = 5, per_facet_k: int = 12, thin_min_results: int = 2, thin_min_similarity: float = 0.15) -> str

Use this when you want to reframe the existing corpus along the axis of a research question.

Projects what you already have (sources, notes, landscape hubs) through a question to show the landscape: it decomposes the question into facets, semantically retrieves the relevant notes per facet, characterizes each facet as established (substantive claims/findings, no recorded contradiction), contested (carries contradicts/qualifies edges), or thin (little relevant material — a gap for this question), and bundles the citations each facet's evidence references. No data is written; this is a read-only lens over the graph.

Parameters:

Name	Type	Description	Default
`question`	`str`	The research question to frame.	required
`project`	`str`	Project name to scope sources; empty frames across everything.	`''`
`facets`	`list[str] \| None`	Optional explicit sub-question/facet strings (agent-decomposed).	`None`
`max_facets`	`int`	Max auto-derived facets when `facets` is not supplied.	`5`
`per_facet_k`	`int`	Notes retrieved per facet.	`12`
`thin_min_results`	`int`	Fewer results than this marks a facet 'thin' (a gap).	`2`
`thin_min_similarity`	`float`	Top similarity below this marks a facet 'thin'.	`0.15`

Source code in zettelkasten/server.py

@mcp_server.tool()
def frame_question(
    question: str,
    project: str = "",
    facets: list[str] | None = None,
    max_facets: int = 5,
    per_facet_k: int = 12,
    thin_min_results: int = 2,
    thin_min_similarity: float = 0.15,
) -> str:
    """Use this when you want to reframe the existing corpus along the axis of a research question.

    Projects what you already have (sources, notes, landscape hubs) through a
    question to show the landscape: it decomposes the question into facets,
    semantically retrieves the relevant notes per facet, characterizes each
    facet as **established** (substantive claims/findings, no recorded
    contradiction), **contested** (carries contradicts/qualifies edges), or
    **thin** (little relevant material — a gap for this question), and bundles
    the citations each facet's evidence references. No data is written; this is
    a read-only lens over the graph.

    Facets come from one of two places:
      - if `facets` is given, those strings are used (let the agent decompose
        the question into sub-questions and pass them here); otherwise
      - facets are derived from the existing landscape `concept` hubs most
        relevant to the question — i.e. the field's own decomposition.

    Args:
        question: The research question to frame.
        project: Project name to scope sources; empty frames across everything.
        facets: Optional explicit sub-question/facet strings (agent-decomposed).
        max_facets: Max auto-derived facets when `facets` is not supplied.
        per_facet_k: Notes retrieved per facet.
        thin_min_results: Fewer results than this marks a facet 'thin' (a gap).
        thin_min_similarity: Top similarity below this marks a facet 'thin'.
    """
    from zettelkasten import framing

    result = framing.frame_question(
        question, get_graph=_get_graph, project=project, facets=facets,
        max_facets=max_facets, per_facet_k=per_facet_k,
        thin_min_results=thin_min_results, thin_min_similarity=thin_min_similarity,
    )
    return json.dumps(result)

guide ¶

guide(action: Literal['apply', 'next'] = 'apply', project: str = '', graph: str = '', params: dict | None = None) -> str

Use this when navigating conditional requirement notes as a decision tree (codes/standards/assumptions).

A normative document (a building code, an engineering standard, a quant model's assumptions) extracted under the reference-standard schema is a graph of requirement notes, each optionally carrying an applies_when block ({condition, predicate?}). Given a design parameter context this walks that graph, prunes branches whose predicates are definitively false, and returns the applicable clauses with their grounding quotes + citations — the compact, cited bundle a simulator feeds itself instead of re-reading the source. Read-only; nothing is written.

Scope the walk with EITHER graph (one source/synthesis graph name) OR project (its sources walked as a forest). params is the design context, e.g. {"seismic_category": "D", "height_m": 32, "material": "steel"}; predicates are ANDed and compared with a whitelisted evaluator (>= > <= < == != in / not in) — never arbitrary code. A node is unresolved when it has a human condition but no machine predicate, or a predicate referencing a param absent from params.

Actions — name(required, optional?): apply(project?, graph?, params?): [default] the full pruned bundle — nodes (each kept node with its evidence + children uids), roots, and flat applicable/unresolved/excluded uid lists. next(project?, graph?, params?): the shallowest UNRESOLVED decision for an interactive step-through; its missing_params name what to supply.

Provide either project or graph. Read-only; all return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def guide(
    action: Literal["apply", "next"] = "apply",
    project: str = "",
    graph: str = "",
    params: dict | None = None,
) -> str:
    """Use this when navigating conditional ``requirement`` notes as a decision tree (codes/standards/assumptions).

    A normative document (a building code, an engineering standard, a quant
    model's assumptions) extracted under the ``reference-standard`` schema is a
    graph of ``requirement`` notes, each optionally carrying an ``applies_when``
    block (``{condition, predicate?}``). Given a design parameter context this
    walks that graph, prunes branches whose predicates are definitively false,
    and returns the applicable clauses with their grounding quotes + citations —
    the compact, cited bundle a simulator feeds itself instead of re-reading the
    source. Read-only; nothing is written.

    Scope the walk with EITHER ``graph`` (one source/synthesis graph name) OR
    ``project`` (its ``sources`` walked as a forest). ``params`` is the design
    context, e.g. ``{"seismic_category": "D", "height_m": 32, "material":
    "steel"}``; predicates are ANDed and compared with a whitelisted evaluator
    (``>= > <= < == != in / not in``) — never arbitrary code. A node is
    ``unresolved`` when it has a human ``condition`` but no machine predicate, or a
    predicate referencing a param absent from ``params``.

    Actions — name(required, optional?):
        apply(project?, graph?, params?): [default] the full pruned bundle —
            ``nodes`` (each kept node with its ``evidence`` + ``children`` uids),
            ``roots``, and flat ``applicable``/``unresolved``/``excluded`` uid lists.
        next(project?, graph?, params?): the shallowest UNRESOLVED decision for an
            interactive step-through; its ``missing_params`` name what to supply.

    Provide either ``project`` or ``graph``. Read-only; all return JSON.
    """
    from zettelkasten import decision_tree

    if not (project or graph):
        return json.dumps({
            "error": "Provide either 'project' or 'graph' to scope the decision tree.",
            "type": "BadRequest",
        })
    if action not in ("apply", "next"):
        return _bad_action("guide", action)

    result = decision_tree.walk(
        graph=graph, params=params or {}, project=project, get_graph=_get_graph,
    )
    if action == "next":
        nxt = decision_tree.next_decision(result)
        return json.dumps({
            "next": nxt,
            "params": result["params"],
            "graphs": result["graphs"],
            "remaining_unresolved": len(result["unresolved"]),
            "summary": result["summary"],
        })
    return json.dumps(result)

synapse ¶

synapse(action: Literal['search', 'frame', 'get', 'connections', 'matrix', 'synthesize', 'debate', 'navigate', 'build'], query: str = '', question: str = '', node_id: str = '', projects: list[str] | None = None, top_k: int = 15, max_nodes: int = -1, hops: int = -1, trace_floor: float = -1.0, token_budget: int = -1, render_max_body_chars: int = -1, expand: bool = False, max_sources: int = -1, facets: list[str] | None = None, max_facets: int = 5, per_facet_k: int = 12, thin_min_results: int = 2, thin_min_similarity: float = 0.15, min_confidence: float = -1.0, synthesize: bool = False, max_rows: int = 60, max_columns: int = 60, per_node_k: int | None = None, sim_threshold: float = 0.45, max_pairs: int = 200, kind: str = 'generic', overfetch: int | None = None, model: str = '') -> str

Use this when working across stores: the layer linking the memory tree (.memory/, episodic "practice") and the zettelkasten (propositional "canon").

Read-only except build (which writes only synapse's own overlay, never .memory/ or .zettelkasten/). Scope is the intersecting ZK projects: omit projects for the configured intersection, pass a list to override, or [] for memory only. Hits are tagged store (memory|zk) and tier (practice|canon). search/frame/build/navigate run OUT-OF-PROCESS to contain native-kglite SIGSEGVs (mapped to a clean NativeCrash); SYNAPSE_INPROCESS_READ=1 forces the in-process read path (search/frame only).

Actions — name(required, optional?): search(query, projects?, top_k?, expand?, max_sources?): hybrid RRF-fused search across both stores; expand also pulls each hit's cross-store neighbours. frame(question, projects?, facets?, max_facets?, per_facet_k?, thin_min_results?, thin_min_similarity?, max_sources?): frame a research question into facets (established/contested/thin). get(node_id): read the FULL body of a MEMORY entry (for a ZK note use get_note). connections(node_id, min_confidence?): typed cross-store edges touching a node. matrix(projects?, min_confidence?, synthesize?, max_rows?, max_columns?, kind?): project the overlay as a practice × canon matrix; synthesize adds LLM row/column/apex summaries; kind="claim" uses the claim-aligned overlay. synthesize(projects?, min_confidence?, max_rows?, max_columns?): grounded per-cell briefs over the CLAIM overlay (cites memory ids + retrieved quotes, invents nothing). debate(projects?): the claim debate map fed with cross-store counter-evidence (practice nodes that contradict a canon claim). navigate(question, projects?, max_nodes?, hops?, trace_floor?, token_budget?, render_max_body_chars?, model?): run the grounded cross-store navigation loop — assemble a neighborhood for question, then let the deterministic Navigator steer an LLM to an ANSWER/ABSTAIN verdict it can never fabricate. READ-ONLY (no CANONICAL-store writes — .memory/, .zettelkasten/, the link overlay, .synapse/ — though, like search/frame, the disposable .angelo/ embedding cache may still be written/warmed on a cold assembly) and runs OUT-OF-PROCESS (the agent bridge and cold kglite assembly cannot run in the stdio server; a native crash maps to a clean NativeCrash). Any assembly or model failure degrades to a labeled abstain. max_nodes=-1 uses the default neighborhood cap, 0 removes the caller-specified count cap while retaining a token-derived pre-assembly safety ceiling, and >0 sets an explicit hard cap shared by assembly and Navigator expansion. hops>0 enables pre-loop neighborhood expansion; trace_floor tunes the router's prose gate. token_budget raises/lowers the navigator's hard per-run token guardrail and render_max_body_chars caps the SHOWN body length (a positive value elides long bodies to a lossy view the ground move re-fetches in full, so a body-heavy neighborhood no longer halts on budget before the model is called; 0 disables elision). render_max_body_chars bounds per-node body DEPTH, NOT neighborhood WIDTH — a wide hub (many neighbors) is bounded by token_budget / max_nodes, never by this knob. Each defers to the navigator default when unset (negative). model requests a Cursor model id; the bridge forwards it without claiming that the provider exposed a resolved runtime model. Returns the verdict, wire status, reasons, halted_reason, conclusion, serialized envelope, and hop/renavigation telemetry. Accepted answers additionally carry a deterministic versioned answer_surface containing display text, epistemic status, resolved citation addresses, and inseparable outdated disclosure. Every abstention/failure carries answer_surface: null. build(projects?, per_node_k?, sim_threshold?, min_confidence?, max_pairs?, kind?, overfetch?): (re)build the connection overlay (semantic NN + shared provenance, LLM-typed) to .synapse/links/. WRITE, refused when write-free. kind="claim" builds the claim-aligned overlay (overfetch widens recall).

max_sources=-1 (default) uses the synapse config; 0 disables source pruning; a positive value caps it (memory always in scope). All return JSON.

Source code in zettelkasten/server.py

@mcp_server.tool()
def synapse(
    action: Literal[
        "search", "frame", "get", "connections", "matrix", "synthesize", "debate",
        "navigate", "build",
    ],
    query: str = "",
    question: str = "",
    node_id: str = "",
    projects: list[str] | None = None,
    top_k: int = 15,
    max_nodes: int = -1,
    hops: int = -1,
    trace_floor: float = -1.0,
    token_budget: int = -1,
    render_max_body_chars: int = -1,
    expand: bool = False,
    max_sources: int = -1,
    facets: list[str] | None = None,
    max_facets: int = 5,
    per_facet_k: int = 12,
    thin_min_results: int = 2,
    thin_min_similarity: float = 0.15,
    min_confidence: float = -1.0,
    synthesize: bool = False,
    max_rows: int = 60,
    max_columns: int = 60,
    per_node_k: int | None = None,
    sim_threshold: float = 0.45,
    max_pairs: int = 200,
    kind: str = "generic",
    overfetch: int | None = None,
    model: str = "",
) -> str:
    """Use this when working across stores: the layer linking the memory tree (``.memory/``, episodic "practice") and the zettelkasten (propositional "canon").

    Read-only except ``build`` (which writes only synapse's own overlay, never
    ``.memory/`` or ``.zettelkasten/``). Scope is the intersecting ZK projects:
    omit ``projects`` for the configured intersection, pass a list to override, or
    ``[]`` for memory only. Hits are tagged ``store`` (memory|zk) and ``tier``
    (practice|canon). ``search``/``frame``/``build``/``navigate`` run
    OUT-OF-PROCESS to contain native-kglite SIGSEGVs (mapped to a clean
    ``NativeCrash``); ``SYNAPSE_INPROCESS_READ=1`` forces the in-process read
    path (search/frame only).

    Actions — name(required, optional?):
        search(query, projects?, top_k?, expand?, max_sources?): hybrid RRF-fused
            search across both stores; ``expand`` also pulls each hit's cross-store
            neighbours.
        frame(question, projects?, facets?, max_facets?, per_facet_k?, thin_min_results?, thin_min_similarity?, max_sources?):
            frame a research question into facets (established/contested/thin).
        get(node_id): read the FULL body of a MEMORY entry (for a ZK note use
            ``get_note``).
        connections(node_id, min_confidence?): typed cross-store edges touching a
            node.
        matrix(projects?, min_confidence?, synthesize?, max_rows?, max_columns?, kind?):
            project the overlay as a practice × canon matrix; ``synthesize`` adds
            LLM row/column/apex summaries; ``kind="claim"`` uses the claim-aligned
            overlay.
        synthesize(projects?, min_confidence?, max_rows?, max_columns?): grounded
            per-cell briefs over the CLAIM overlay (cites memory ids + retrieved
            quotes, invents nothing).
        debate(projects?): the claim debate map fed with cross-store
            counter-evidence (practice nodes that contradict a canon claim).
        navigate(question, projects?, max_nodes?, hops?, trace_floor?, token_budget?, render_max_body_chars?, model?): run the
            grounded cross-store navigation loop — assemble a neighborhood for
            ``question``, then let the deterministic Navigator steer an LLM to an
            ANSWER/ABSTAIN verdict it can never fabricate. READ-ONLY (no
            CANONICAL-store writes — ``.memory/``, ``.zettelkasten/``, the link
            overlay, ``.synapse/`` — though, like ``search``/``frame``, the
            disposable ``.angelo/`` embedding cache may still be written/warmed on
            a cold assembly) and runs OUT-OF-PROCESS (the agent bridge and cold
            kglite assembly cannot run in the stdio server; a native crash maps to
            a clean ``NativeCrash``). Any assembly or model failure degrades to a
            labeled abstain. ``max_nodes=-1`` uses the default neighborhood cap,
            ``0`` removes the caller-specified count cap while retaining a
            token-derived pre-assembly safety ceiling, and ``>0`` sets an explicit
            hard cap shared by assembly and Navigator expansion. ``hops>0`` enables
            pre-loop neighborhood expansion;
            ``trace_floor`` tunes the router's prose gate. ``token_budget`` raises/lowers the navigator's
            hard per-run token guardrail and ``render_max_body_chars`` caps the
            SHOWN body length (a positive value elides long bodies to a lossy view
            the ``ground`` move re-fetches in full, so a body-heavy neighborhood
            no longer halts on budget before the model is called; ``0`` disables
            elision). ``render_max_body_chars`` bounds per-node body DEPTH, NOT
            neighborhood WIDTH — a wide hub (many neighbors) is bounded by
            ``token_budget`` / ``max_nodes``, never by this knob. Each defers to
            the navigator default when unset (negative). ``model`` requests a
            Cursor model id; the bridge forwards it without claiming that the
            provider exposed a resolved runtime model.
            Returns the verdict, wire status, reasons, ``halted_reason``,
            conclusion, serialized envelope, and hop/renavigation telemetry.
            Accepted answers additionally carry a deterministic versioned
            ``answer_surface`` containing display text, epistemic status, resolved
            citation addresses, and inseparable outdated disclosure. Every
            abstention/failure carries ``answer_surface: null``.
        build(projects?, per_node_k?, sim_threshold?, min_confidence?, max_pairs?, kind?, overfetch?):
            (re)build the connection overlay (semantic NN + shared provenance,
            LLM-typed) to ``.synapse/links/``. WRITE, refused when write-free.
            ``kind="claim"`` builds the claim-aligned overlay (``overfetch`` widens
            recall).

    ``max_sources=-1`` (default) uses the synapse config; ``0`` disables source
    pruning; a positive value caps it (memory always in scope). All return JSON.
    """
    action = (action or "").strip().lower()
    projects_n = _synapse_projects(projects)
    # -1 (the default) => defer to the synapse config for the source fan-out cap;
    # 0 disables pruning; a positive value caps it. Passed as ``None`` when
    # deferring so retrieval/the subprocess consults the config.
    max_sources_n = None if max_sources < 0 else max_sources

    if action == "search":
        if not query.strip():
            return json.dumps({"error": "query must not be empty.", "type": "ValidationError"})
        if _synapse_read_inprocess():
            from zettelkasten.synapse import retrieval
            return json.dumps(retrieval.search_across_stores(
                query, _get_graph, projects=projects_n, top_k=top_k, expand=expand,
                max_sources=max_sources_n,
            ))
        # Default: run out-of-process. Cross-store search fans out to every
        # in-scope embedding index (incl. always-cold federated peers), whose
        # first-access native kglite indexing has SIGSEGV'd — which would take
        # this stdio MCP server down. The subprocess contains the crash and a
        # faulthandler trace localizes it. See _run_synapse_query_subprocess.
        return _run_synapse_query({
            "action": "search", "query": query, "projects": projects_n,
            "top_k": top_k, "expand": expand, "max_sources": max_sources_n,
        })
    if action == "frame":
        if not question.strip():
            return json.dumps({
                "error": "question must not be empty.",
                "type": "ValidationError",
                "answer_surface": None,
            })
        if _synapse_read_inprocess():
            from zettelkasten.synapse import retrieval
            return json.dumps(retrieval.frame_across_stores(
                question, _get_graph, projects=projects_n, facets=facets,
                max_facets=max_facets, per_facet_k=per_facet_k,
                thin_min_results=thin_min_results, thin_min_similarity=thin_min_similarity,
                max_sources=max_sources_n,
            ))
        return _run_synapse_query({
            "action": "frame", "question": question, "projects": projects_n,
            "facets": facets, "max_facets": max_facets, "per_facet_k": per_facet_k,
            "thin_min_results": thin_min_results, "thin_min_similarity": thin_min_similarity,
            "max_sources": max_sources_n,
        })
    if action == "get":
        from zettelkasten.synapse.memory_source import get_memory_source
        note = get_memory_source().notes.get(node_id.strip())
        if note is None:
            return json.dumps({
                "error": f"'{node_id}' is not a memory entry in this repo. "
                         f"For a zettelkasten note, use get_note.",
                "type": "NotFound",
            })
        e = note.entry
        return json.dumps({
            "id": note.id, "store": "memory", "tier": "practice", "type": note.type,
            "title": note.title, "tags": note.tags, "body": note.body,
            "parent_id": e.get("parent_id"), "status": e.get("status"),
            "created_at": e.get("created_at"), "updated_at": e.get("updated_at"),
            "files": e.get("files"), "related_to": e.get("related_to"),
        })
    if action == "connections":
        from zettelkasten.synapse import overlay as _overlay
        mc = min_confidence if min_confidence >= 0 else 0.0
        nid = node_id.strip()
        overlay_kind = "claim" if kind == "claim" else "generic"
        edges = [e for e in _overlay.get_connections(nid, kind=overlay_kind)
                 if float(e.get("confidence", 0.0)) >= mc]
        return json.dumps({"node_id": nid, "connections": edges, "count": len(edges)})
    if action == "matrix":
        from zettelkasten.synapse import matrix as _matrix
        from zettelkasten.synapse import overlay as _overlay
        mc = min_confidence if min_confidence >= 0 else 0.0
        overlay_kind = "claim" if kind == "claim" else "generic"
        lens = _matrix.build_matrix_lens(
            _get_graph, projects=projects_n, min_confidence=mc,
            max_rows=max_rows, max_columns=max_columns,
            overlay=_overlay.load_overlay(overlay_kind),
        )
        if synthesize:
            lens = _matrix.synthesize_axes(lens)
        return json.dumps(lens)
    if action == "synthesize":
        # Grounded per-cell briefs over the CLAIM overlay. Read-only + injectable;
        # the LLM synth degrades to "" on failure (mirrors matrix synthesize_axes),
        # so this is safe in-process.
        from zettelkasten.synapse import matrix as _matrix
        from zettelkasten.synapse import overlay as _overlay
        from zettelkasten.synapse import synthesis as _synthesis
        mc = min_confidence if min_confidence >= 0 else 0.0
        lens = _matrix.build_matrix_lens(
            _get_graph, projects=projects_n, min_confidence=mc,
            max_rows=max_rows, max_columns=max_columns,
            overlay=_overlay.load_overlay(_synthesis.CLAIM_OVERLAY_KIND),
        )
        material_fn = _synthesis.claim_material_provider(_get_graph, projects=projects_n)
        lens = _synthesis.synthesize_matrix(lens, material_fn)
        return json.dumps(lens)
    if action == "debate":
        from zettelkasten import claims as _claims
        from zettelkasten.synapse import synthesis as _synthesis
        mc = min_confidence if min_confidence >= 0 else 0.55
        cross = _synthesis.cross_store_contradictions(min_confidence=mc)
        return json.dumps(_claims.debate_map(
            _get_graph, project="", graph="",
            cross_store_contradictions=cross,
        ))
    if action == "navigate":
        # Grounded cross-store navigation loop. Runs OUT-OF-PROCESS: it drives the
        # cursor-sdk agent bridge (which cannot come up inside this stdio MCP
        # server or the warm read worker — see build) and cold-assembles native
        # kglite indexes that can SIGSEGV. The child contains the crash and prints
        # a single result line; a Python-level failure degrades to a labeled
        # abstain inside the child (read-only, enforced-honesty verdict). See
        # _run_synapse_navigate_subprocess / zettelkasten.synapse.navigate.
        if not question.strip():
            return json.dumps({
                "error": "question must not be empty.",
                "type": "ValidationError",
                "answer_surface": None,
            })
        from zettelkasten.synapse.navigate import (
            NavigationValidationError,
            validate_navigation_parameters,
            validation_abstention,
        )
        try:
            validate_navigation_parameters(
                max_nodes=max_nodes,
                hops=hops,
                token_budget=token_budget,
                render_max_body_chars=render_max_body_chars,
            )
        except NavigationValidationError as exc:
            return json.dumps(validation_abstention(str(exc)))
        return _run_synapse_navigate_subprocess({
            "question": question, "projects": projects_n,
            "max_nodes": max_nodes, "hops": hops, "trace_floor": trace_floor,
            "token_budget": token_budget,
            "render_max_body_chars": render_max_body_chars,
            "model": model,
        })
    if action == "build":
        blocked = _writes_blocked()
        if blocked:
            return blocked
        mc = min_confidence if min_confidence >= 0 else 0.55
        build_kind = "claim" if kind == "claim" else "generic"
        # ``per_node_k``/``overfetch`` default to None so a caller can request the
        # generic 5/1 explicitly without it being mistaken for "unset". An unset
        # dial resolves to the kind's default: the claim path overfetches and raises
        # per_node_k (CLAIM_* dials) to offset the cross-register type filter.
        if build_kind == "claim":
            from zettelkasten.synapse import synthesis as _synthesis
            pnk = per_node_k if per_node_k is not None else _synthesis.CLAIM_PER_NODE_K
            ofetch = overfetch if overfetch is not None else _synthesis.CLAIM_OVERFETCH
        else:
            pnk = per_node_k if per_node_k is not None else 5
            ofetch = overfetch if overfetch is not None else 1
        # Runs out-of-process: the LLM typing bridge cannot come up inside this
        # stdio MCP server (its stdin/stdout are the JSON-RPC transport), so the
        # build is spawned as a fresh `python -m zettelkasten.synapse.build`.
        return _run_synapse_build_subprocess(
            projects_n, pnk, sim_threshold, mc, max_pairs, kind=build_kind, overfetch=ofetch,
        )

    return _bad_action("synapse", action)

zettelkasten.server¶