Author SHA1 Message Date
TotenKnopfandClaude Fable 5.1 d9bd4596ed chore: untrack IDE files
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-06 20:40:24 +02:00
TotenKnopfandClaude Fable 5.1 96780317a7 refactor: extract kb_framework package, build graph in memory
Scripts become thin CLIs over kb_framework (graph, query, query_tools,
vault, instructions) so the hosted read-only connector shares one
implementation with the stdio server. The MCP server builds the graph
from Markdown on demand and invalidates it on writes; refresh_graph
rebuilds in memory and reports validation problems. Pre-commit hook
validates only; index/graph.json is no longer read, written or staged
(generate-graph.py --write remains for operators). Golden fixtures pin
graph and tool payloads to the previous output. Agent and skill docs
updated to match.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-10-06 20:40:13 +02:00
TotenKnopf 4fec637599 revert .gitignore changes 2026-09-20 16:44:41 +00:00
TotenKnopf 293b582e97 Support external vault roots across tooling 2026-09-20 17:05:21 +02:00
35 changed files with 2325 additions and 1039 deletions
+10
View File
@@ -0,0 +1,10 @@
.git
.idea
.venv
__pycache__
tests
.claude
index
skills
agents
0*_*/
+1 -1
View File
@@ -1,3 +1,3 @@
__pycache__/
*.pyc
.claude/settings.local.json
.claude/settings.local.json.idea/
+29 -12
View File
@@ -4,12 +4,27 @@ title: Agent Query Guide
type: meta
status: active
tags: [agents, jq, graph]
last_reviewed: 2026-06-16
last_reviewed: 2026-10-06
---
# Agent Query Guide
`index/graph.json` is for navigation. You can query tags or types there. Use it!
The graph is the selected vault's navigation index, built in memory from the
Markdown by `kb_framework.graph.build_graph` — no file is read or committed.
MCP is the canonical cross-repository interface: `graph_summary`,
`graph_search_nodes`, `graph_neighbors`, and `read_note`; the server rebuilds
the graph after every write, and `refresh_graph(validate=true)` rebuilds it on
demand and reports validation problems.
MCP-only agents use the tools above; shell examples below are for operators,
not agent fallback. For those queries, explicitly set `KNOWLEDGE_BASE_ROOT` and change into
that directory. Framework commands require `KNOWLEDGE_BASE_FRAMEWORK_ROOT` as
an explicit path; skill or agent file locations do not identify the data vault.
```bash
: "${KNOWLEDGE_BASE_ROOT:?Set the data vault directory explicitly}"
cd -- "$KNOWLEDGE_BASE_ROOT"
```
Schema:
@@ -22,7 +37,7 @@ Schema:
- `id` (string): stable note identifier (usually from frontmatter)
- `title` (string): note title
- `path` (string): vault-relative markdown path
- `repo` (string): repo scope (`tenant-controller`, `controller-util`, `global`, ...)
- `repo` (string): repo scope (a repository identifier, or `global`)
- `type` (string): note classification from a controlled vocabulary —
`hub|project|area|guideline|policy|feature|roadmap|decision|note|meta|reference`;
absent frontmatter means `note`, and `--validate` fails on anything else
@@ -58,9 +73,11 @@ optional JSON brace suffix carrying the relationship and annotation:
see-also, source`.
- Legacy `[[wiki links]]` are **no longer accepted**: `generate-graph.py --validate`
fails on any that remain. Locate and convert them with
`python3 scripts/list-wiki-links.py`.
`python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/list-wiki-links.py" --vault-root "$KNOWLEDGE_BASE_ROOT"`.
Examples with `jq`:
Examples with `jq` (operators only; first dump the graph with
`python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --write`,
which writes `index/graph.json` — keep `index/` out of git):
```bash
# list repos
@@ -74,17 +91,17 @@ jq -r '[.nodes[].tags[]?] | group_by(.) | map({tag: .[0], count: length})
jq -r '.nodes | sort_by(-(.inbound_count // 0))[] | "\(.inbound_count // 0)\t\(.id)\t\(.path)"' index/graph.json
# notes in one repo
jq -r '.nodes[] | select(.repo=="tenant-controller") | "\(.id)\t\(.path)"' index/graph.json
jq -r '.nodes[] | select(.repo=="<repo>") | "\(.id)\t\(.path)"' index/graph.json
# all projects (PARA category 1)
jq -r '.nodes[] | select(.type=="project") | "\(.id)\t\(.path)"' index/graph.json
# outgoing links from one note
NOTE_ID="repo/controller-util/info"
NOTE_ID="<note-id>"
jq -r --arg id "$NOTE_ID" '.nodes[] | select(.id==$id) | .outbound_links[]?' index/graph.json
# incoming links to one note
NOTE_ID="repo/controller-util/info"
NOTE_ID="<note-id>"
jq -r --arg id "$NOTE_ID" '.edges[] | select(.to==$id) | .from' index/graph.json | sort -u
# outbound edges of a note WITH relationship type + annotation
@@ -104,11 +121,11 @@ Validation:
```bash
# validate: non-zero exit on unresolved links, edge types or note types
# outside their vocabularies, OR any remaining legacy [[wiki links]]
python3 scripts/generate-graph.py --validate
python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --validate
# regenerate graph and validate
python3 scripts/generate-graph.py && python3 scripts/generate-graph.py --validate
# write a graph dump for jq AND validate in one run
python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/generate-graph.py" --vault-root "$KNOWLEDGE_BASE_ROOT" --write --validate
# list any remaining legacy [[wiki links]] (rejected by --validate) to convert
python3 scripts/list-wiki-links.py
python3 "$KNOWLEDGE_BASE_FRAMEWORK_ROOT/scripts/list-wiki-links.py" --vault-root "$KNOWLEDGE_BASE_ROOT"
```
+9 -7
View File
@@ -1,4 +1,6 @@
# Run framework tools and install integrations against the selected data vault.
PYTHON ?= python3
VAULT_ROOT ?= $(if $(KNOWLEDGE_BASE_ROOT),$(KNOWLEDGE_BASE_ROOT),$(CURDIR))
CLAUDE_HOME ?= $(HOME)/.claude
MCP_NAME ?= knowledge-database
@@ -8,18 +10,18 @@ test:
$(PYTHON) -m unittest discover -s tests
generate-graph:
$(PYTHON) scripts/generate-graph.py
$(PYTHON) scripts/generate-graph.py --vault-root "$(VAULT_ROOT)"
validate-graph:
$(PYTHON) scripts/generate-graph.py --validate
$(PYTHON) scripts/generate-graph.py --vault-root "$(VAULT_ROOT)" --validate
graph: generate-graph validate-graph
# `graph` is the everyday gate: validate only, nothing written.
graph: validate-graph
install-hooks:
chmod +x scripts/pre-commit
ln -sf ../../scripts/pre-commit .git/hooks/pre-commit
$(PYTHON) scripts/install-hooks.py --vault-root "$(VAULT_ROOT)"
# Link this vault's skills and agents into the global Claude context
# Link this framework's skills and agents into the global Claude context
# (~/.claude/skills, ~/.claude/agents) so they are invocable from any repo.
install-global-claude:
mkdir -p "$(CLAUDE_HOME)/skills" "$(CLAUDE_HOME)/agents"
@@ -40,5 +42,5 @@ install-global-claude:
# so it is available from any repo). Re-run to update; removes stale entry first.
install-mcp-claude:
-claude mcp remove --scope user "$(MCP_NAME)" 2>/dev/null
claude mcp add --scope user "$(MCP_NAME)" -- $(PYTHON) "$(CURDIR)/scripts/mcp-server.py"
claude mcp add --scope user "$(MCP_NAME)" -- $(PYTHON) "$(CURDIR)/scripts/mcp-server.py" --vault-root "$(VAULT_ROOT)"
@echo "mcp -> $(MCP_NAME) ($(CURDIR)/scripts/mcp-server.py)"
+56 -6
View File
@@ -24,7 +24,7 @@ writing notes.
02_areas/ ongoing responsibilities & standards (policies, value streams, decisions)
03_resources/ reference material (repo info, ingested externals)
04_archives/ inactive items moved out of the three live categories
index/ generated graph (index/graph.json)
kb_framework/ importable core: in-memory graph build, queries, note access policy
scripts/ graph generator, link inventory, MCP server, git hook
agents/ subagent definitions, symlinked into .claude/agents/
skills/ skill definitions, symlinked into .claude/skills/
@@ -37,20 +37,70 @@ folder with an `index.md`.
## Bootstrap
```sh
make install-hooks # pre-commit hook: regenerate + validate the graph
make install-hooks # pre-commit hook: validate the graph (nothing written)
make install-global-claude # link skills & agents into ~/.claude so any repo can use them
make install-mcp-claude # register scripts/mcp-server.py as the knowledge-database MCP server
make graph # generate index/graph.json and validate it
make graph # validate the graph (built in memory); `make generate-graph` writes index/graph.json
make test # run the tooling tests
```
## External data vault
Keep this repository as the control plane (tools, skills, agents, hooks, tests,
operational documentation), and store notes in a separate data repository.
The empty PARA skeleton also supports self-contained use.
All Python commands select data with `--vault-root`, then the
`KNOWLEDGE_BASE_ROOT` environment variable, then this framework's root.
An explicit root must be an existing directory; Git operations require it to
be the Git repository root. The working directory does not select the vault.
```sh
python3 /path/to/framework/scripts/mcp-server.py --vault-root /path/to/vault
make -C /path/to/framework graph VAULT_ROOT=/path/to/vault
make -C /path/to/framework install-hooks VAULT_ROOT=/path/to/vault
make -C /path/to/framework install-mcp-claude VAULT_ROOT=/path/to/vault
```
`VAULT_ROOT` defaults to `KNOWLEDGE_BASE_ROOT`, then the framework directory.
The hook installer honors Git's hooks directory (including worktrees and
`core.hooksPath`), preserves unmanaged hooks, and writes a wrapper pointing at
framework tooling. Reinstall after moving the framework or vault. The hook
validates the selected vault's graph in memory; it writes nothing and never
requires a staged `index/graph.json`.
Configure each stdio MCP client with command `python3` and arguments
`["/path/to/framework/scripts/mcp-server.py", "--vault-root", "/path/to/vault"]`.
MCP reads, writes and Git operations use that vault. The graph is built in
memory on first use and rebuilt after every write; `refresh_graph` rebuilds it
on demand and, with `validate=true`, applies the commit gate. No graph file is
read or written. A README resource is advertised only if present.
Global skill and agent symlinks must point into this framework. MCP is their
only interface for vault access; unavailable operations are reported without
shell fallback. The CLI examples here are for operators. No personal taxonomy is assumed.
Agents call `read_instructions` to load framework `AGENTS.md` and
`AGENT_QUERY_GUIDE.md` even before a data graph exists. This read-only tool
returns those two framework documents; `read_note` remains vault-scoped.
For migration, test the external root first, back up MCP configurations and
symlink targets, cut clients and hooks over, then smoke-test initialize,
tools/list, graph_summary, search, read and refresh. Only remove old tool copies
after this passes. On failure restore the configs and symlinks and retain the
old copies. Restart existing clients to load the new server command.
A minimal data repository contains `01_projects/` through `04_archives/`
and repository hygiene; `index/` is not committed. Before removing a root hub, replace
its inbound links with links between PARA hubs and validate the graph. Preserve
untracked files separately from the tracked-file cleanup.
## Generation
- Run `make graph`
- Output is written to `index/graph.json`
- The graph is derived in memory by `kb_framework.graph.build_graph`; run `make graph` to validate it
- `make generate-graph` writes `index/graph.json` for operators who want a file; keep `index/` out of git
## Rules
- Markdown is the source of truth
- `index/graph.json` is derived and generated
- The graph is derived; never hand-edit or commit a graph dump
- Keep notes linked using typed markdown links: `[text](note-id){"type":"…"}` (legacy `[[wiki links]]` are rejected by `make validate-graph`)
- Organize by PARA actionability; move notes between buckets as that changes
- Fix broken links — find them with `make graph` or `make validate-graph`
+18 -20
View File
@@ -21,25 +21,28 @@ external URLs that notes cite — never to reach the vault itself.
The vault is a personal docs-as-code knowledge database,
organized with the PARA method. Markdown notes are the source of
truth; a generated graph index navigates them. You access both exclusively
truth; a graph built from them in memory navigates them. You access both exclusively
through MCP.
## MCP tools
- `read_instructions` — framework `AGENTS.md` and `AGENT_QUERY_GUIDE.md`.
Call this first and read both; report failure if unavailable. These documents
live outside the data graph, and the server locates them without a guessed path.
- `graph_summary` — overview of the graph: counts, repo scopes, the tag
vocabulary (`by_tag`, most frequent first) and central nodes.
- `graph_search_nodes` — find nodes by tag / type / repo / id / text.
- `graph_neighbors` — outbound + inbound links for a node id, each with its edge
`type` + `annotation` (traverse the graph).
- `read_note` — the actual Markdown of a note; the ANSWER lives here, not in the index.
- `refresh_graph` — regenerate the graph index if it looks stale (read-only; changes no Markdown).
- `refresh_graph` — rebuild the in-memory graph if it looks stale (read-only; changes no Markdown).
## Mandate
Given a **topic or task**, return a complete, cited briefing:
1. **Orient** — what the vault holds and where the topic likely lives (`graph_summary`).
2. **Find** — every relevant note, via `graph_search_nodes` (+ text-search fallback).
2. **Find** — every relevant note, via `graph_search_nodes`, then `read_note` for candidate note bodies.
3. **Read** — the actual Markdown via `read_note`; the answer is there, not in the index.
4. **Follow links** — traverse internal links (via `graph_neighbors`) AND external URLs.
5. **Synthesize** — one briefing, every claim cited to a note path/id.
@@ -48,14 +51,14 @@ Given a **topic or task**, return a complete, cited briefing:
```
01_projects/ active efforts with a goal and an end
02_areas/ ongoing standards: policies, value-stream areas, decisions (ADRs)
03_resources/meta/ taxonomy + note templates
03_resources/repos/<repo>/ per-repo info.md (+ decisions/)
03_resources/repos/repo-landscape.md index of documented repos
02_areas/ ongoing responsibilities and standards
03_resources/ reference material
04_archives/ inactive items
README.md vault hub
```
Framework owns agents, skills and tooling. The MCP server selects the data vault;
its directory is independent of this agent definition's location.
Every note carries taxonomy frontmatter: `id`, `title`, `type`
(`hub|project|area|guideline|policy|feature|roadmap|decision|note|meta|reference`, enforced by
`generate-graph.py --validate`), `status`
@@ -68,23 +71,18 @@ Legacy `[[wiki links]]` are rejected by the graph validator and must not be writ
Start broad with `graph_summary` to see node counts, repo scopes, the tag
vocabulary and the most-central notes. Its `by_tag` list is the only place the
tag spellings are visible — read it before filtering by `tag`, which matches
verbatim. Then `read_note` the entry points:
- vault hub (`README.md`) — entry points.
- `03_resources/repos/repo-landscape.md` — index of documented repos, grouped.
- `02_areas/ways-of-working.md` — global engineering policy.
- `03_resources/meta/taxonomy.md` — frontmatter fields and PARA buckets.
For a repo topic, `read_note` on `03_resources/repos/<repo>/info.md`. If
`graph_search_nodes` returns no node for it, say it is not yet documented.
verbatim. Find hubs with `graph_search_nodes(type="hub")` and read relevant
ones with `read_note`. Discover policies and repo scopes through the graph;
do not assume a root README or any particular taxonomy subfolders exist.
If no relevant node exists, say the topic is not yet documented.
## 2. Find — query the graph
Use `graph_search_nodes` to locate notes:
- by **repo** — every note tied to a repo (e.g. `tenant-controller`).
- by **repo** — every note tied to a repo (e.g. `<repo>`).
- by **type** — `policy`, `decision`, `feature`, `guideline`, `project`, ...
- by **tag** — e.g. `kcp`. Exact match; take the spelling from `by_tag`.
- by **tag** — e.g. a value from `by_tag`. Exact match; take the spelling from `by_tag`.
- by **text** — `query` matches a node's id, title, path and tags, not the note
body. Widen with a tag or repo filter rather than a longer phrase.
@@ -119,7 +117,7 @@ Links are the intended navigation. Traverse, do not guess.
- Answer the topic/task directly from what the notes say.
- Cite every claim by vault-relative `path` (e.g.
`03_resources/repos/controller-util/info.md`) and/or `id`.
`03_resources/<resource>/index.md`) and/or `id`.
- Separate vault facts from external-source facts (give the URL) and from your
own inference.
- Flag `status: draft`/`deprecated` notes. If the vault does not cover part of
+26 -29
View File
@@ -1,6 +1,6 @@
---
name: the-scribe
description: MCP-only resource ingestor for the knowledge-base vault (PARA-organized). Hand it a URL, a local file path, or a PDF and it runs the `knowledge-base-ingest-resource` pipeline entirely through the `knowledge-database` MCP server — captures the original bytes (text or binary) into `raw/`, writes the resource note and one note per topic, wires typed markdown links, regenerates + validates the graph, then commits the named paths and pushes. Has no shell. Use when the user says "ingest", "add this resource", or "save this link/paper" and wants the result written and published.
description: MCP-only resource ingestor for the knowledge-base vault (PARA-organized). Hand it a URL, a local file path, or a PDF and it runs the `knowledge-base-ingest-resource` pipeline entirely through the `knowledge-database` MCP server — captures the original bytes (text or binary) into `raw/`, writes the resource note and one note per topic, wires typed markdown links, rebuilds + validates the graph, then commits the named paths and pushes. Has no shell. Use when the user says "ingest", "add this resource", or "save this link/paper" and wants the result written and published.
tools: Read, WebFetch, Skill, mcp__knowledge-database__*
mcpServers:
- knowledge-database
@@ -25,6 +25,8 @@ already captured into `raw/` (a PDF, an image) at the absolute path
## MCP tools
Read:
- `read_instructions` — framework `AGENTS.md` and `AGENT_QUERY_GUIDE.md`,
available without a graph or a data-vault copy of either document.
- `graph_summary` — counts, repo scopes, the tag vocabulary (`by_tag`), most-linked notes.
- `graph_search_nodes` — find notes by tag / type / repo / id / free text.
- `graph_neighbors` — inbound + outbound links of a node, with edge `type` and `annotation`.
@@ -42,8 +44,9 @@ Write:
- `edit_note(path, old_string, new_string, replace_all?)` — exact-string edit,
for backlinks into notes that already exist.
Publish:
- `refresh_graph(validate=true)` — regenerate and validate `index/graph.json`.
- `refresh_graph(validate=true)` — rebuild the in-memory graph and validate it; `problems` lists what to fix.
Publish only when the caller explicitly authorizes it:
- `git_commit(message, paths[])` — stage the named paths and commit.
- `git_push(remote?, branch?)` — push to `origin` / current branch.
@@ -55,21 +58,14 @@ Invoke the skill first, then follow it step by step:
Skill(knowledge-base-ingest-resource)
```
The skill is the source of truth for the procedure. Translate each of its shell
steps into the MCP equivalent:
Then call `read_instructions` and read both documents before any vault mutation.
The MCP server locates the framework; no filesystem access or path guessing is
needed. If instructions cannot be loaded, report the failure before writing.
Use the skill's explicit graph queries to discover additional vault policies.
Apply task-specific instructions supplied by the caller as well.
| Skill step | Your call |
|-------------------------------------|--------------------------------------------------|
| check slug not taken / find matches | `graph_search_nodes`, `read_note` |
| `printf … > raw/source.url` | `write_note("03_resources/<slug>/raw/source.url", …)` |
| save fetched page to `raw/page.md` | `WebFetch` → `write_note(".../raw/page.md", …)` |
| `cp "<source-path>" raw/` | `capture_raw(path=".../raw/<file>", source_path="<source-path>")` |
| `curl -fsSL "<url>" -o raw/<file>` | `capture_raw(path=".../raw/<file>", url="<url>")` |
| read the captured binary | `Read(<absolute_path from capture_raw>)` |
| write the topic + index notes | `write_note` |
| backlink an existing note | `edit_note` |
| `make graph` | `refresh_graph(validate=true)` |
| `git add <path>` (one by one) | `git_commit(message, paths=[…])` |
The skill is the source of truth for the procedure. Follow its MCP calls directly;
do not translate missing operations into shell commands.
## Operating rules
@@ -84,25 +80,26 @@ steps into the MCP equivalent:
Never fabricate a page or transcript — if the fetch or capture fails, stop and
report what is missing. Verify the capture landed: the result carries `bytes`
and `sha256`.
- **YouTube is still out of reach.** Transcripts need `yt-dlp` or
`youtube-transcript-api`, which are shell tools you do not have. Capture the
URL into `raw/source.url`, then hand the video to **the-collector** rather than
inventing transcript text.
- **YouTube transcripts must be supplied.** Capture a supplied transcript with
`capture_raw` or `write_note`. If none is supplied, preserve the source URL and
report the missing transcript before writing content notes. Do not run shell
extractors or invent transcript text.
- **Link, don't duplicate.** Search the graph before writing a topic note. If the
vault already covers the topic, link it instead of creating a second note.
- **Typed markdown links only.** `[title](note-id){"type":"…","link-info":"…"}`.
`type` is from the controlled vocabulary — `related, supersedes,
superseded-by, depends-on, part-of, parent, derived-from, see-also, source`.
Legacy `[[wiki links]]` fail validation.
- **Never write the graph.** `index/graph.json` is derived; `write_note` refuses
it. Use `refresh_graph`. Fix every unresolved link until validation is clean —
the pre-commit hook rejects a stale or invalid graph.
- **Never write a graph file.** The graph is derived in memory; `write_note`
refuses `index/graph.json`. Use `refresh_graph`. Fix every unresolved link until
validation is clean — the pre-commit hook rejects an invalid graph.
- **Name every path you commit.** `git_commit` takes an explicit list; wildcard
staging is rejected by the server. Include the resource folder, every
backlinked note, and `index/graph.json`.
- **Push straight to `main`.** This vault takes ingests as direct commits; no
branch, no PR. Commit with Conventional Commits (`feat: ingest <resource>`),
then `git_push`. If the caller says "don't push", stop after the commit.
staging is rejected by the server. Include the resource folder and every
backlinked note; there is no graph file to commit.
- **Follow the caller's publishing instructions.** Do not commit or push unless
explicitly authorized. Use the selected vault's branch and review policy;
do not assume a branch name or a direct-push workflow. Without publishing
authorization, report the changed paths and validation result.
## Report
+10
View File
@@ -0,0 +1,10 @@
"""kb_framework: the importable core of the knowledge-base framework.
Exists so the stdio MCP server (`scripts/mcp-server.py`) and the hosted
read-only connector share one implementation of graph building, graph
queries, note access policy and framework instructions. The graph is built
in memory from the vault's Markdown; nothing here writes to disk.
"""
__all__ = ["graph", "query", "query_tools", "vault", "instructions"]
__version__ = "0.2.0"
+481
View File
@@ -0,0 +1,481 @@
"""Build and validate the vault graph in memory.
One node per Markdown note (frontmatter plus link counts), one edge per typed
markdown link. `build_graph` walks a vault root and returns a `GraphBuild`; it
never touches `index/graph.json`. `validation_problems` is the gate the
pre-commit hook and `--validate` apply: unresolved targets, edge relationship
types or note `type` values outside their controlled vocabularies, and legacy
`[[wiki links]]` are all rejected.
"""
from __future__ import annotations
import json
import re
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Tuple
# Obsidian-style wiki link: [[target#anchor|alias]]
WIKI_LINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|([^\]]+))?\]\]")
# Standard markdown link: [display](target); (?<!!) excludes ![img](...)
MD_LINK_RE = re.compile(r"(?<!!)\[([^\]]*)\]\(([^)]+)\)")
# Trailing "— gloss" (em-dash / en-dash / hyphen) used as an annotation fallback
GLOSS_RE = re.compile(r"^\s*[—–-]\s+(\S.*)$")
FENCED_CODE_RE = re.compile(r"```.*?```", re.DOTALL)
INLINE_CODE_RE = re.compile(r"`[^`\n]+`")
# Markdown targets that are external (not vault edges, not unresolved-link errors)
EXTERNAL_TARGET_RE = re.compile(r"^(?:[a-z][a-z0-9+.\-]*://|mailto:|tel:|#)", re.IGNORECASE)
# Strict controlled vocabulary for edge relationship types. A link with no brace
# (or no "type") defaults to "related"; --validate fails on any type not in this set.
RELATION_TYPES = {
"related",
"supersedes",
"superseded-by",
"depends-on",
"part-of",
"parent",
"derived-from",
"see-also",
"source",
}
# Strict controlled vocabulary for a note's frontmatter `type`. A note without
# one is a "note"; --validate fails on any value outside this set, which is what
# keeps the facet meaningful as the vault grows.
NOTE_TYPES = {
"hub",
"project",
"area",
"guideline",
"policy",
"feature",
"roadmap",
"decision",
"note",
"meta",
"reference",
}
@dataclass
class LinkRef:
target: str # raw target text (resolution key: id / rel-path / title / stem, or a URL)
anchor: str # display / alias text -> edge anchor_text
kind: str # "wiki_link" | "markdown_link"
rel_type: str # semantic relationship type (default "related")
annotation: str # per-edge "what to expect there" note
lineno: int # 1-based line number in the note body
raw: str # the raw link text as written (for the inventory script)
is_external: bool # markdown link whose target is a URL/anchor -> no edge
@dataclass
class Note:
path: Path
rel_path: str
stem: str
meta: Dict[str, object]
title: str
note_id: str
links: List[LinkRef]
@dataclass(frozen=True)
class UnresolvedLink:
from_id: str
from_path: str
target: str
@dataclass(frozen=True)
class WikiLinkUse:
from_id: str
from_path: str
target: str
lineno: int
raw: str
def parse_frontmatter(text: str) -> Tuple[Dict[str, object], str]:
if not text.startswith("---\n"):
return {}, text
end = text.find("\n---\n", 4)
if end == -1:
return {}, text
fm_block = text[4:end]
body = text[end + 5 :]
meta: Dict[str, object] = {}
for raw in fm_block.splitlines():
line = raw.strip()
if not line or line.startswith("#") or ":" not in line:
continue
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
if value.startswith("[") and value.endswith("]"):
inner = value[1:-1].strip()
if inner:
meta[key] = [v.strip().strip('"').strip("'") for v in inner.split(",")]
else:
meta[key] = []
else:
meta[key] = value.strip('"').strip("'")
return meta, body
def derive_title(meta: Dict[str, object], body: str, stem: str) -> str:
if isinstance(meta.get("title"), str) and meta["title"].strip():
return str(meta["title"]).strip()
for line in body.splitlines():
if line.startswith("# "):
return line[2:].strip()
return stem.replace("-", " ").strip().title()
def mask_code(body: str) -> str:
"""Blank out code so links inside code samples are ignored, preserving line
boundaries so line-based parsing (gloss fallback, line numbers) stays aligned."""
body = FENCED_CODE_RE.sub(lambda m: "\n" * m.group(0).count("\n"), body)
body = INLINE_CODE_RE.sub(lambda m: " " * len(m.group(0)), body)
return body
def scan_brace(text: str, start: int) -> Tuple[object, int]:
"""Given text[start] == '{', scan to the matching '}' respecting JSON string
quoting, and json.loads the block. Returns (obj_or_None, end_index_after_brace).
On an unterminated or malformed brace, returns (None, start) so the caller
treats it as no brace."""
depth = 0
in_str = False
esc = False
i = start
while i < len(text):
c = text[i]
if in_str:
if esc:
esc = False
elif c == "\\":
esc = True
elif c == '"':
in_str = False
elif c == '"':
in_str = True
elif c == "{":
depth += 1
elif c == "}":
depth -= 1
if depth == 0:
raw = text[start : i + 1]
try:
return json.loads(raw), i + 1
except (json.JSONDecodeError, ValueError):
return None, start
i += 1
return None, start
def clean_md_target(raw: str) -> str:
"""Normalize a markdown link target: drop a trailing ``"title"`` and <> wrappers."""
raw = raw.strip()
if raw.startswith("<") and raw.endswith(">"):
raw = raw[1:-1].strip()
# `[text](target "title")` -> target
for sep in (' "', " '"):
if sep in raw:
raw = raw.split(sep, 1)[0].strip()
return raw
def parse_links(body: str) -> List[LinkRef]:
masked = mask_code(body)
refs: List[LinkRef] = []
for lineno, line in enumerate(masked.splitlines(), start=1):
# Collect wiki + markdown link matches on this line, in order.
raw_matches: List[Tuple[int, int, str, str, str, str]] = []
for m in WIKI_LINK_RE.finditer(line):
target = m.group(1).strip()
alias = (m.group(2) or target).strip()
if target:
raw_matches.append(
(m.start(), m.end(), "wiki_link", target, alias, m.group(0))
)
for m in MD_LINK_RE.finditer(line):
target = clean_md_target(m.group(2))
display = m.group(1).strip()
if target:
raw_matches.append(
(
m.start(),
m.end(),
"markdown_link",
target,
display or target,
m.group(0),
)
)
raw_matches.sort(key=lambda x: x[0])
line_refs: List[Tuple[LinkRef, int]] = [] # (ref, end-index-after-brace)
for start, end, kind, target, anchor, raw in raw_matches:
rel_type, annotation = "related", ""
brace_end = end
if end < len(line) and line[end] == "{":
obj, be = scan_brace(line, end)
if isinstance(obj, dict):
brace_end = be
t = obj.get("type")
if isinstance(t, str) and t.strip():
rel_type = t.strip()
info = obj.get("link-info")
if isinstance(info, str) and info.strip():
annotation = info.strip()
is_external = kind == "markdown_link" and bool(
EXTERNAL_TARGET_RE.match(target)
)
ref = LinkRef(
target=target,
anchor=anchor,
kind=kind,
rel_type=rel_type,
annotation=annotation,
lineno=lineno,
raw=raw,
is_external=is_external,
)
line_refs.append((ref, brace_end))
# Annotation fallback: a trailing "— gloss" applies only when the link is
# the sole link on its line (so we don't mis-assign a shared gloss).
if len(line_refs) == 1:
ref, brace_end = line_refs[0]
if not ref.annotation:
gloss = GLOSS_RE.match(line[brace_end:])
if gloss:
ref.annotation = gloss.group(1).strip()
refs.extend(ref for ref, _ in line_refs)
return refs
def norm(s: str) -> str:
return s.strip().lower()
def build_notes(root: Path) -> List[Note]:
notes: List[Note] = []
for p in sorted(root.rglob("*.md")):
rel = p.relative_to(root)
if rel.parts and rel.parts[0] in {"index", "scripts", ".obsidian"}:
continue
# `raw/` holds untouched ingested originals (external READMEs, specs, …),
# not vault notes — skip them so their file-relative markdown links are not
# treated as (broken) graph edges.
if "raw" in rel.parts:
continue
text = p.read_text(encoding="utf-8")
meta, body = parse_frontmatter(text)
title = derive_title(meta, body, p.stem)
note_id = str(meta.get("id") or rel.with_suffix("")).strip()
links = parse_links(body)
# parse_links numbers lines within `body`; shift to absolute file line
# numbers by the number of lines the frontmatter block occupied.
line_offset = text.count("\n", 0, len(text) - len(body))
if line_offset:
for ref in links:
ref.lineno += line_offset
notes.append(
Note(
path=p,
rel_path=rel.as_posix(),
stem=p.stem,
meta=meta,
title=title,
note_id=note_id,
links=links,
)
)
return notes
def build_alias_map(notes: List[Note]) -> Dict[str, str]:
alias_map: Dict[str, str] = {}
stem_buckets: Dict[str, List[str]] = defaultdict(list)
for n in notes:
stem_buckets[norm(n.stem)].append(n.note_id)
for n in notes:
alias_map[norm(n.note_id)] = n.note_id
alias_map[norm(n.rel_path)] = n.note_id
alias_map[norm(Path(n.rel_path).with_suffix("").as_posix())] = n.note_id
alias_map[norm(n.title)] = n.note_id
for stem, ids in stem_buckets.items():
if len(ids) == 1:
alias_map[stem] = ids[0]
return alias_map
def generate_graph(
root: Path, existing_generated_at: str = ""
) -> Tuple[Dict[str, object], List[UnresolvedLink], List[WikiLinkUse]]:
notes = build_notes(root)
alias_map = build_alias_map(notes)
edges_set = set()
outbound: Dict[str, List[str]] = defaultdict(list)
inbound_count: Dict[str, int] = defaultdict(int)
unresolved_set = set()
wiki_uses_set = set()
for n in notes:
for ref in n.links:
# Legacy [[wiki links]] are still parsed (so we can flag them) but are
# no longer an accepted link form — --validate rejects any that remain.
if ref.kind == "wiki_link":
wiki_uses_set.add(
WikiLinkUse(
from_id=n.note_id,
from_path=n.rel_path,
target=ref.target,
lineno=ref.lineno,
raw=ref.raw,
)
)
if ref.is_external:
continue
target = alias_map.get(norm(ref.target))
if not target:
unresolved_set.add(
UnresolvedLink(
from_id=n.note_id,
from_path=n.rel_path,
target=ref.target,
)
)
continue
edge = (n.note_id, target, ref.kind, ref.rel_type, ref.annotation, ref.anchor)
if edge in edges_set:
continue
edges_set.add(edge)
outbound[n.note_id].append(target)
inbound_count[target] += 1
nodes = []
for n in sorted(notes, key=lambda x: x.note_id):
tags = n.meta.get("tags", [])
if not isinstance(tags, list):
tags = []
nodes.append(
{
"id": n.note_id,
"title": n.title,
"path": n.rel_path,
"repo": n.meta.get("repo", "global"),
"type": n.meta.get("type", "note"),
"status": n.meta.get("status", "active"),
"tags": tags,
"last_reviewed": n.meta.get("last_reviewed", ""),
"outbound_links": sorted(set(outbound.get(n.note_id, []))),
"inbound_count": inbound_count.get(n.note_id, 0),
}
)
edges = [
{
"from": src,
"to": dst,
"kind": kind,
"type": rel_type,
"anchor_text": anchor,
"annotation": annotation,
"weight": 1.0,
}
for src, dst, kind, rel_type, annotation, anchor in sorted(edges_set)
]
graph = {
"version": 1,
"generated_at": existing_generated_at
if existing_generated_at
else datetime.now(timezone.utc).isoformat(),
"nodes": nodes,
"edges": edges,
}
unresolved = sorted(unresolved_set, key=lambda x: (x.from_path, x.target))
wiki_uses = sorted(wiki_uses_set, key=lambda x: (x.from_path, x.lineno, x.raw))
return graph, unresolved, wiki_uses
def find_invalid_types(graph: Dict[str, object]) -> List[Dict[str, object]]:
return [
e
for e in graph.get("edges", []) # type: ignore[union-attr]
if e.get("type") not in RELATION_TYPES
]
def find_invalid_node_types(graph: Dict[str, object]) -> List[Dict[str, object]]:
return [
n
for n in graph.get("nodes", []) # type: ignore[union-attr]
if n.get("type") not in NOTE_TYPES
]
@dataclass
class GraphBuild:
"""Result of one in-memory build: the graph plus everything --validate reports."""
graph: Dict[str, object]
unresolved: List[UnresolvedLink] = field(default_factory=list)
wiki_uses: List[WikiLinkUse] = field(default_factory=list)
@property
def invalid_edge_types(self) -> List[Dict[str, object]]:
return find_invalid_types(self.graph)
@property
def invalid_node_types(self) -> List[Dict[str, object]]:
return find_invalid_node_types(self.graph)
@property
def ok(self) -> bool:
return not validation_problems(self)
def build_graph(root: Path, existing_generated_at: str = "") -> GraphBuild:
"""Build the graph for `root` in memory."""
graph, unresolved, wiki_uses = generate_graph(Path(root), existing_generated_at)
return GraphBuild(graph=graph, unresolved=unresolved, wiki_uses=wiki_uses)
def validation_problems(build: GraphBuild) -> List[str]:
"""Human-readable validation failures, empty when the build is clean."""
problems: List[str] = []
for link in build.unresolved:
problems.append(f"unresolved link: {link.from_path}: [[{link.target}]]")
for e in build.invalid_edge_types:
problems.append(
f"unknown relationship type: {e.get('from')} -> {e.get('to')}: type={e.get('type')!r}"
)
for n in build.invalid_node_types:
problems.append(f"unknown note type: {n.get('id')} ({n.get('path')}): type={n.get('type')!r}")
for w in build.wiki_uses:
problems.append(f"legacy wiki link: {w.from_path}:{w.lineno}: {w.raw}")
return problems
+35
View File
@@ -0,0 +1,35 @@
"""Locate and read the framework's agent instructions.
`read_instructions` returns AGENTS.md and AGENT_QUERY_GUIDE.md so an agent
whose only access is MCP still gets the operating rules. In a source checkout
the documents live next to this package (the framework root); in an installed
wheel they are bundled under `kb_framework/_docs/`. Both lookups are explicit —
there is never a fallback to files inside the data vault.
"""
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict
DOCUMENT_NAMES = ("AGENTS.md", "AGENT_QUERY_GUIDE.md")
_PACKAGE_DIR = Path(__file__).resolve().parent
# Source checkout: the framework repository root (parent of this package).
FRAMEWORK_ROOT = _PACKAGE_DIR.parent
_BUNDLED_DOCS = _PACKAGE_DIR / "_docs"
def instructions_root() -> Path:
"""Directory holding the instruction documents for this installation."""
if (FRAMEWORK_ROOT / DOCUMENT_NAMES[0]).is_file():
return FRAMEWORK_ROOT
return _BUNDLED_DOCS
def read_instructions() -> Dict[str, Any]:
root = instructions_root()
documents = [
{"path": name, "content": (root / name).read_text(encoding="utf-8")}
for name in DOCUMENT_NAMES
]
return {"framework_root": str(root), "documents": documents}
+159
View File
@@ -0,0 +1,159 @@
"""Pure query functions over an in-memory graph dict.
These back the read-only MCP tools (`graph_summary`, `graph_search_nodes`,
`graph_neighbors`) and the id-to-path lookup `read_note` needs. They take the
graph as data and never read the filesystem, so the stdio server and the
connector return identical payloads for identical graphs.
"""
from __future__ import annotations
from collections import Counter
from typing import Any, Dict, List
# graph_summary defaults: enough most-linked notes to spot the hubs, and enough
# tags to cover the vocabulary an agent actually filters on.
DEFAULT_TOP_LINKED = 10
DEFAULT_TOP_TAGS = 50
DEFAULT_SEARCH_LIMIT = 30
DEFAULT_NEIGHBOR_LIMIT = 50
def nodes_by_id(graph: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
nodes = graph.get("nodes", [])
return {str(node.get("id")): node for node in nodes if isinstance(node, dict)}
def match_filters(node: Dict[str, Any], args: Dict[str, Any]) -> bool:
repo = args.get("repo")
note_type = args.get("type")
status = args.get("status")
tag = args.get("tag")
query = str(args.get("query", "")).strip().lower()
if repo and node.get("repo") != repo:
return False
if note_type and node.get("type") != note_type:
return False
if status and node.get("status") != status:
return False
if tag and tag not in (node.get("tags") or []):
return False
if query:
haystacks = [
str(node.get("id", "")).lower(),
str(node.get("title", "")).lower(),
str(node.get("path", "")).lower(),
" ".join(str(t).lower() for t in (node.get("tags") or [])),
]
if not any(query in h for h in haystacks):
return False
return True
def summary(
graph: Dict[str, Any],
topLinked: int = DEFAULT_TOP_LINKED,
topTags: int = DEFAULT_TOP_TAGS,
) -> Dict[str, Any]:
nodes = graph.get("nodes", [])
edges = graph.get("edges", [])
top_n = int(topLinked)
top_tags = int(topTags)
by_repo: Counter[str] = Counter(str(node.get("repo", "unknown")) for node in nodes)
tag_counts: Counter[str] = Counter(
str(tag) for node in nodes for tag in (node.get("tags") or [])
)
# Frequent tags first, alphabetical within a tie, so a truncated list is
# both the most useful slice and stable across calls.
ranked_tags = sorted(tag_counts.items(), key=lambda item: (-item[1], item[0]))
sorted_nodes = sorted(
nodes,
key=lambda n: int(n.get("inbound_count", 0)),
reverse=True,
)[:top_n]
return {
"version": graph.get("version"),
"generated_at": graph.get("generated_at"),
"node_count": len(nodes),
"edge_count": len(edges),
"by_repo": dict(by_repo),
"tag_count": len(tag_counts),
"by_tag": [
{"tag": tag, "count": count} for tag, count in ranked_tags[:top_tags]
],
"top_linked": [
{
"id": node.get("id"),
"title": node.get("title"),
"path": node.get("path"),
"inbound_count": node.get("inbound_count", 0),
}
for node in sorted_nodes
],
}
def search_nodes(graph: Dict[str, Any], args: Dict[str, Any]) -> Dict[str, Any]:
nodes = graph.get("nodes", [])
limit = int(args.get("limit", DEFAULT_SEARCH_LIMIT))
filtered = [node for node in nodes if match_filters(node, args)]
filtered.sort(key=lambda n: str(n.get("id", "")))
sliced = filtered[:limit]
return {
"total_matches": len(filtered),
"returned": len(sliced),
"items": sliced,
}
def neighbors(
graph: Dict[str, Any], id: str, limit: int = DEFAULT_NEIGHBOR_LIMIT
) -> Dict[str, Any]:
node_id = str(id)
limit = int(limit)
nodes = nodes_by_id(graph)
if node_id not in nodes:
raise ValueError(f"node '{node_id}' not found")
edges = graph.get("edges", [])
outbound_edges = [e for e in edges if e.get("from") == node_id]
inbound_edges = [e for e in edges if e.get("to") == node_id]
def describe(edge_list: List[Dict[str, Any]], direction: str) -> List[Dict[str, Any]]:
# neighbor id is the "other end" relative to node_id
other_key = "to" if direction == "outbound" else "from"
result: List[Dict[str, Any]] = []
for edge in edge_list:
item_id = str(edge.get(other_key))
node = nodes.get(item_id)
result.append(
{
"id": node.get("id") if node else item_id,
"title": node.get("title") if node else "<missing>",
"path": node.get("path") if node else "",
"direction": direction,
"type": edge.get("type", "related"),
"anchor_text": edge.get("anchor_text", ""),
"annotation": edge.get("annotation", ""),
}
)
return result[:limit]
return {
"node": nodes[node_id],
"inbound_count": len(inbound_edges),
"outbound_count": len(outbound_edges),
"inbound": describe(inbound_edges, "inbound"),
"outbound": describe(outbound_edges, "outbound"),
}
def node_path(graph: Dict[str, Any], id: str) -> str:
"""Vault-relative path of the note behind a node id."""
node = nodes_by_id(graph).get(str(id))
if not node:
raise ValueError(f"node '{id}' not found")
return str(node.get("path"))
+164
View File
@@ -0,0 +1,164 @@
"""The read-only MCP tool contract: definitions and dispatch.
Both front ends (the stdio server and the hosted connector) advertise exactly
these five tools with exactly these schemas, and `call` produces the result
dict (`content` + `structuredContent`) each returns. Keeping the contract here
means an agent skill written against one front end stays valid on the other.
"""
from __future__ import annotations
import json
from pathlib import Path
from typing import Any, Dict, List
from kb_framework import instructions, query, vault
READ_ONLY_TOOL_NAMES = (
"read_instructions",
"graph_summary",
"graph_search_nodes",
"graph_neighbors",
"read_note",
)
def definitions() -> List[Dict[str, Any]]:
return [
{
"name": "read_instructions",
"description": (
"Read framework AGENTS.md and AGENT_QUERY_GUIDE.md before working "
"with the vault. These operating documents live outside an "
"external vault and do not require a generated graph."
),
"inputSchema": {
"type": "object",
"properties": {},
"additionalProperties": False,
},
},
{
"name": "graph_summary",
"description": (
"Orientation for a cold start: node and edge counts, the repo "
"scopes, the most-linked notes, and the tag vocabulary as "
"`by_tag` (most frequent first) plus `tag_count`, the number of "
"distinct tags. Read `by_tag` before filtering "
"`graph_search_nodes` by `tag` — that filter matches a tag "
"verbatim, so a guessed spelling returns nothing."
),
"inputSchema": {
"type": "object",
"properties": {
"topLinked": {
"type": "integer",
"minimum": 1,
"maximum": 50,
"description": "Most-linked notes to return (default 10).",
},
"topTags": {
"type": "integer",
"minimum": 1,
"maximum": 500,
"description": "Tags to return, most frequent first (default 50).",
},
},
"additionalProperties": False,
},
},
{
"name": "graph_search_nodes",
"description": "Search graph nodes by metadata and free text.",
"inputSchema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"type": {"type": "string"},
"status": {"type": "string"},
"tag": {"type": "string"},
"query": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 200},
},
},
},
{
"name": "graph_neighbors",
"description": (
"Show inbound and outbound links for a node ID. Each neighbor "
"includes the edge `type` (relationship, e.g. supersedes, "
"depends-on), `anchor_text` (the link's display text), and "
"`annotation` (what to expect in that note)."
),
"inputSchema": {
"type": "object",
"required": ["id"],
"properties": {
"id": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 200},
},
},
},
{
"name": "read_note",
"description": (
"Read note content by graph node ID or vault-relative path. "
"Provide exactly one of `id` or `path`. "
"If both or neither are provided, the call returns an error."
),
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "string", "description": "Graph node ID (e.g. 'AGENTS')."},
"path": {"type": "string", "description": "Vault-relative file path (e.g. 'README.md')."},
},
"additionalProperties": False,
},
},
]
def text_result(payload: Dict[str, Any]) -> Dict[str, Any]:
"""Wrap a payload as an MCP tool result: pretty JSON text plus structuredContent."""
return {
"content": [{"type": "text", "text": json.dumps(payload, indent=2)}],
"structuredContent": payload,
}
def read_note_path(graph: Dict[str, Any], args: Dict[str, Any]) -> str:
"""Resolve `read_note` arguments to the vault-relative path to read."""
has_id = bool(args.get("id"))
has_path = bool(args.get("path"))
if has_id and has_path:
raise ValueError("provide exactly one of 'id' or 'path', not both")
if not has_id and not has_path:
raise ValueError("exactly one of 'id' or 'path' is required")
if has_id:
return query.node_path(graph, str(args["id"]))
return str(args["path"])
def call(
name: str, args: Dict[str, Any], *, graph: Dict[str, Any], root: Path
) -> Dict[str, Any]:
"""Run one read-only tool. Raises ValueError for unknown tools or bad input."""
if name == "read_instructions":
return text_result(instructions.read_instructions())
if name == "graph_summary":
return text_result(
query.summary(
graph,
topLinked=args.get("topLinked", query.DEFAULT_TOP_LINKED),
topTags=args.get("topTags", query.DEFAULT_TOP_TAGS),
)
)
if name == "graph_search_nodes":
return text_result(query.search_nodes(graph, args))
if name == "graph_neighbors":
return text_result(
query.neighbors(graph, str(args["id"]), args.get("limit", query.DEFAULT_NEIGHBOR_LIMIT))
)
if name == "read_note":
return text_result(vault.read_note_payload(root, read_note_path(graph, args)))
raise ValueError(f"unknown tool '{name}'")
+57
View File
@@ -0,0 +1,57 @@
"""Filesystem access to a vault, confined to its root.
`resolve_in_root` is the single place that turns a vault-relative path into an
absolute one and refuses empty, absolute or escaping values, so no tool can
reach outside the vault. `read_note_payload` returns the `read_note` payload
shape shared by the stdio server and the connector: UTF-8 text inline, or a
binary descriptor (size, sha256, hint) for captured originals under `raw/`.
"""
from __future__ import annotations
import hashlib
from pathlib import Path
from typing import Any, Dict
def resolve_in_root(root: Path, raw_path: Any, *, label: str = "path") -> Path:
"""Turn a vault-relative path into an absolute path inside `root`."""
value = str(raw_path or "").strip()
if not value:
raise ValueError(f"'{label}' is required")
candidate = Path(value)
if candidate.is_absolute():
raise ValueError(f"'{label}' must be vault-relative, got absolute '{value}'")
root = Path(root).resolve()
target = (root / candidate).resolve()
if target != root and not target.is_relative_to(root):
raise ValueError(f"'{label}' resolves outside repository root: '{value}'")
return target
def relative_to_root(root: Path, target: Path) -> str:
return target.relative_to(Path(root).resolve()).as_posix()
def read_note_payload(root: Path, raw_path: Any) -> Dict[str, Any]:
"""The `read_note` payload for a vault-relative path."""
target = resolve_in_root(root, raw_path)
if not target.exists():
raise FileNotFoundError(f"note not found at {target}")
relative = relative_to_root(root, target)
data = target.read_bytes()
try:
text = data.decode("utf-8")
except UnicodeDecodeError:
return {
"path": relative,
"binary": True,
"bytes": len(data),
"sha256": hashlib.sha256(data).hexdigest(),
"hint": (
f"{target} is not UTF-8 text (e.g. a PDF or image captured under "
"raw/). Open it with a file-reading tool that handles binaries; "
"this server returns text only."
),
}
return {"path": relative, "binary": False, "content": text}
+24
View File
@@ -0,0 +1,24 @@
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "kb-framework"
version = "0.2.0"
description = "Knowledge-base framework core: in-memory graph, query tools, vault access policy"
readme = "README.md"
requires-python = ">=3.12"
license = { text = "MIT" }
dependencies = []
[tool.hatch.build.targets.wheel]
packages = ["kb_framework"]
# Bundle the agent instructions so `read_instructions` works from an installed
# wheel (e.g. inside the hosted connector image), not only from a checkout.
[tool.hatch.build.targets.wheel.force-include]
"AGENTS.md" = "kb_framework/_docs/AGENTS.md"
"AGENT_QUERY_GUIDE.md" = "kb_framework/_docs/AGENT_QUERY_GUIDE.md"
[tool.hatch.build.targets.sdist]
include = ["kb_framework", "AGENTS.md", "AGENT_QUERY_GUIDE.md", "README.md", "pyproject.toml"]
+57 -467
View File
@@ -1,451 +1,64 @@
#!/usr/bin/env python3
"""Build `index/graph.json` from the vault's Markdown, and validate it.
"""Validate the vault graph, and optionally write it to `index/graph.json`.
The graph is the derived navigation layer agents query instead of walking the
tree: one node per note (frontmatter plus link counts) and one edge per typed
markdown link. `--validate` is the gate that keeps it trustworthy — it fails on
links that resolve to nothing, on edge relationship types or note `type` values
outside their controlled vocabularies, and on legacy `[[wiki links]]`.
Thin CLI over `kb_framework.graph`: the graph is built in memory from the
vault's Markdown. `--validate` is the gate the pre-commit hook applies — it
fails on links that resolve to nothing, on edge relationship types or note
`type` values outside their controlled vocabularies, and on legacy
`[[wiki links]]`. `--write` is for operators who want the file; nothing else
writes it, and the MCP servers never read it.
"""
import argparse
import json
import re
import sys
from collections import defaultdict
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Dict, List, Tuple
from typing import Dict, List
from vault_root import parse_vault_args
# Obsidian-style wiki link: [[target#anchor|alias]]
WIKI_LINK_RE = re.compile(r"\[\[([^\]|#]+)(?:#[^\]|]+)?(?:\|([^\]]+))?\]\]")
# Standard markdown link: [display](target); (?<!!) excludes ![img](...)
MD_LINK_RE = re.compile(r"(?<!!)\[([^\]]*)\]\(([^)]+)\)")
# Trailing "— gloss" (em-dash / en-dash / hyphen) used as an annotation fallback
GLOSS_RE = re.compile(r"^\s*[—–-]\s+(\S.*)$")
FENCED_CODE_RE = re.compile(r"```.*?```", re.DOTALL)
INLINE_CODE_RE = re.compile(r"`[^`\n]+`")
# Markdown targets that are external (not vault edges, not unresolved-link errors)
EXTERNAL_TARGET_RE = re.compile(r"^(?:[a-z][a-z0-9+.\-]*://|mailto:|tel:|#)", re.IGNORECASE)
# Strict controlled vocabulary for edge relationship types. A link with no brace
# (or no "type") defaults to "related"; --validate fails on any type not in this set.
RELATION_TYPES = {
"related",
"supersedes",
"superseded-by",
"depends-on",
"part-of",
"parent",
"derived-from",
"see-also",
"source",
}
# Strict controlled vocabulary for a note's frontmatter `type`. A note without
# one is a "note"; --validate fails on any value outside this set, which is what
# keeps the facet meaningful as the vault grows.
NOTE_TYPES = {
"hub",
"project",
"area",
"guideline",
"policy",
"feature",
"roadmap",
"decision",
"note",
"meta",
"reference",
}
@dataclass
class LinkRef:
target: str # raw target text (resolution key: id / rel-path / title / stem, or a URL)
anchor: str # display / alias text -> edge anchor_text
kind: str # "wiki_link" | "markdown_link"
rel_type: str # semantic relationship type (default "related")
annotation: str # per-edge "what to expect there" note
lineno: int # 1-based line number in the note body
raw: str # the raw link text as written (for the inventory script)
is_external: bool # markdown link whose target is a URL/anchor -> no edge
@dataclass
class Note:
path: Path
rel_path: str
stem: str
meta: Dict[str, object]
title: str
note_id: str
links: List[LinkRef]
@dataclass(frozen=True)
class UnresolvedLink:
from_id: str
from_path: str
target: str
@dataclass(frozen=True)
class WikiLinkUse:
from_id: str
from_path: str
target: str
lineno: int
raw: str
def parse_frontmatter(text: str) -> Tuple[Dict[str, object], str]:
if not text.startswith("---\n"):
return {}, text
end = text.find("\n---\n", 4)
if end == -1:
return {}, text
fm_block = text[4:end]
body = text[end + 5 :]
meta: Dict[str, object] = {}
for raw in fm_block.splitlines():
line = raw.strip()
if not line or line.startswith("#") or ":" not in line:
continue
key, value = line.split(":", 1)
key = key.strip()
value = value.strip()
if value.startswith("[") and value.endswith("]"):
inner = value[1:-1].strip()
if inner:
meta[key] = [v.strip().strip('"').strip("'") for v in inner.split(",")]
else:
meta[key] = []
else:
meta[key] = value.strip('"').strip("'")
return meta, body
def derive_title(meta: Dict[str, object], body: str, stem: str) -> str:
if isinstance(meta.get("title"), str) and meta["title"].strip():
return str(meta["title"]).strip()
for line in body.splitlines():
if line.startswith("# "):
return line[2:].strip()
return stem.replace("-", " ").strip().title()
def mask_code(body: str) -> str:
"""Blank out code so links inside code samples are ignored, preserving line
boundaries so line-based parsing (gloss fallback, line numbers) stays aligned."""
body = FENCED_CODE_RE.sub(lambda m: "\n" * m.group(0).count("\n"), body)
body = INLINE_CODE_RE.sub(lambda m: " " * len(m.group(0)), body)
return body
def scan_brace(text: str, start: int) -> Tuple[object, int]:
"""Given text[start] == '{', scan to the matching '}' respecting JSON string
quoting, and json.loads the block. Returns (obj_or_None, end_index_after_brace).
On an unterminated or malformed brace, returns (None, start) so the caller
treats it as no brace."""
depth = 0
in_str = False
esc = False
i = start
while i < len(text):
c = text[i]
if in_str:
if esc:
esc = False
elif c == "\\":
esc = True
elif c == '"':
in_str = False
elif c == '"':
in_str = True
elif c == "{":
depth += 1
elif c == "}":
depth -= 1
if depth == 0:
raw = text[start : i + 1]
try:
return json.loads(raw), i + 1
except (json.JSONDecodeError, ValueError):
return None, start
i += 1
return None, start
def clean_md_target(raw: str) -> str:
"""Normalize a markdown link target: drop a trailing ``"title"`` and <> wrappers."""
raw = raw.strip()
if raw.startswith("<") and raw.endswith(">"):
raw = raw[1:-1].strip()
# `[text](target "title")` -> target
for sep in (' "', " '"):
if sep in raw:
raw = raw.split(sep, 1)[0].strip()
return raw
def parse_links(body: str) -> List[LinkRef]:
masked = mask_code(body)
refs: List[LinkRef] = []
for lineno, line in enumerate(masked.splitlines(), start=1):
# Collect wiki + markdown link matches on this line, in order.
raw_matches: List[Tuple[int, int, str, str, str, str]] = []
for m in WIKI_LINK_RE.finditer(line):
target = m.group(1).strip()
alias = (m.group(2) or target).strip()
if target:
raw_matches.append(
(m.start(), m.end(), "wiki_link", target, alias, m.group(0))
)
for m in MD_LINK_RE.finditer(line):
target = clean_md_target(m.group(2))
display = m.group(1).strip()
if target:
raw_matches.append(
(
m.start(),
m.end(),
"markdown_link",
target,
display or target,
m.group(0),
)
)
raw_matches.sort(key=lambda x: x[0])
line_refs: List[Tuple[LinkRef, int]] = [] # (ref, end-index-after-brace)
for start, end, kind, target, anchor, raw in raw_matches:
rel_type, annotation = "related", ""
brace_end = end
if end < len(line) and line[end] == "{":
obj, be = scan_brace(line, end)
if isinstance(obj, dict):
brace_end = be
t = obj.get("type")
if isinstance(t, str) and t.strip():
rel_type = t.strip()
info = obj.get("link-info")
if isinstance(info, str) and info.strip():
annotation = info.strip()
is_external = kind == "markdown_link" and bool(
EXTERNAL_TARGET_RE.match(target)
)
ref = LinkRef(
target=target,
anchor=anchor,
kind=kind,
rel_type=rel_type,
annotation=annotation,
lineno=lineno,
raw=raw,
is_external=is_external,
)
line_refs.append((ref, brace_end))
# Annotation fallback: a trailing "— gloss" applies only when the link is
# the sole link on its line (so we don't mis-assign a shared gloss).
if len(line_refs) == 1:
ref, brace_end = line_refs[0]
if not ref.annotation:
gloss = GLOSS_RE.match(line[brace_end:])
if gloss:
ref.annotation = gloss.group(1).strip()
refs.extend(ref for ref, _ in line_refs)
return refs
def norm(s: str) -> str:
return s.strip().lower()
def build_notes(root: Path) -> List[Note]:
notes: List[Note] = []
for p in sorted(root.rglob("*.md")):
rel = p.relative_to(root)
if rel.parts and rel.parts[0] in {"index", "scripts", ".obsidian"}:
continue
# `raw/` holds untouched ingested originals (external READMEs, specs, …),
# not vault notes — skip them so their file-relative markdown links are not
# treated as (broken) graph edges.
if "raw" in rel.parts:
continue
text = p.read_text(encoding="utf-8")
meta, body = parse_frontmatter(text)
title = derive_title(meta, body, p.stem)
note_id = str(meta.get("id") or rel.with_suffix("")).strip()
links = parse_links(body)
# parse_links numbers lines within `body`; shift to absolute file line
# numbers by the number of lines the frontmatter block occupied.
line_offset = text.count("\n", 0, len(text) - len(body))
if line_offset:
for ref in links:
ref.lineno += line_offset
notes.append(
Note(
path=p,
rel_path=rel.as_posix(),
stem=p.stem,
meta=meta,
title=title,
note_id=note_id,
links=links,
)
)
return notes
def build_alias_map(notes: List[Note]) -> Dict[str, str]:
alias_map: Dict[str, str] = {}
stem_buckets: Dict[str, List[str]] = defaultdict(list)
for n in notes:
stem_buckets[norm(n.stem)].append(n.note_id)
for n in notes:
alias_map[norm(n.note_id)] = n.note_id
alias_map[norm(n.rel_path)] = n.note_id
alias_map[norm(Path(n.rel_path).with_suffix("").as_posix())] = n.note_id
alias_map[norm(n.title)] = n.note_id
for stem, ids in stem_buckets.items():
if len(ids) == 1:
alias_map[stem] = ids[0]
return alias_map
def generate_graph(
root: Path, existing_generated_at: str = ""
) -> Tuple[Dict[str, object], List[UnresolvedLink], List[WikiLinkUse]]:
notes = build_notes(root)
alias_map = build_alias_map(notes)
edges_set = set()
outbound: Dict[str, List[str]] = defaultdict(list)
inbound_count: Dict[str, int] = defaultdict(int)
unresolved_set = set()
wiki_uses_set = set()
for n in notes:
for ref in n.links:
# Legacy [[wiki links]] are still parsed (so we can flag them) but are
# no longer an accepted link form — --validate rejects any that remain.
if ref.kind == "wiki_link":
wiki_uses_set.add(
WikiLinkUse(
from_id=n.note_id,
from_path=n.rel_path,
target=ref.target,
lineno=ref.lineno,
raw=ref.raw,
)
)
if ref.is_external:
continue
target = alias_map.get(norm(ref.target))
if not target:
unresolved_set.add(
UnresolvedLink(
from_id=n.note_id,
from_path=n.rel_path,
target=ref.target,
)
)
continue
edge = (n.note_id, target, ref.kind, ref.rel_type, ref.annotation, ref.anchor)
if edge in edges_set:
continue
edges_set.add(edge)
outbound[n.note_id].append(target)
inbound_count[target] += 1
nodes = []
for n in sorted(notes, key=lambda x: x.note_id):
tags = n.meta.get("tags", [])
if not isinstance(tags, list):
tags = []
nodes.append(
{
"id": n.note_id,
"title": n.title,
"path": n.rel_path,
"repo": n.meta.get("repo", "global"),
"type": n.meta.get("type", "note"),
"status": n.meta.get("status", "active"),
"tags": tags,
"last_reviewed": n.meta.get("last_reviewed", ""),
"outbound_links": sorted(set(outbound.get(n.note_id, []))),
"inbound_count": inbound_count.get(n.note_id, 0),
}
)
edges = [
{
"from": src,
"to": dst,
"kind": kind,
"type": rel_type,
"anchor_text": anchor,
"annotation": annotation,
"weight": 1.0,
}
for src, dst, kind, rel_type, annotation, anchor in sorted(edges_set)
]
graph = {
"version": 1,
"generated_at": existing_generated_at
if existing_generated_at
else datetime.now(timezone.utc).isoformat(),
"nodes": nodes,
"edges": edges,
}
unresolved = sorted(unresolved_set, key=lambda x: (x.from_path, x.target))
wiki_uses = sorted(wiki_uses_set, key=lambda x: (x.from_path, x.lineno, x.raw))
return graph, unresolved, wiki_uses
from kb_framework.graph import ( # noqa: F401 (re-exported for list-wiki-links.py and tests)
NOTE_TYPES,
RELATION_TYPES,
GraphBuild,
LinkRef,
Note,
UnresolvedLink,
WikiLinkUse,
build_alias_map,
build_graph,
build_notes,
find_invalid_node_types,
find_invalid_types,
generate_graph,
norm,
parse_frontmatter,
parse_links,
)
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="Generate and validate Obsidian graph JSON"
description="Validate the vault graph; optionally write index/graph.json"
)
parser.add_argument(
"--validate",
action="store_true",
help="validate the graph: unresolved targets, edge relationship types, "
"note types, "
"and legacy [[wiki links]] (all rejected)",
"note types, and legacy [[wiki links]] (all rejected)",
)
parser.add_argument(
"--write",
action="store_true",
help="write index/graph.json (default unless --validate is used alone)",
)
return parser.parse_args()
return parse_vault_args(parser)
def print_unresolved(unresolved: List[UnresolvedLink]) -> None:
if not unresolved:
print("Validation passed: no unresolved links")
return
print(f"Validation failed: {len(unresolved)} unresolved link(s)")
for link in unresolved:
print(f"- {link.from_path}: [[{link.target}]]")
@@ -455,92 +68,69 @@ def print_wiki_links(wiki_uses: List[WikiLinkUse]) -> None:
if not wiki_uses:
print("Validation passed: no legacy [[wiki links]]")
return
print(f"Validation failed: {len(wiki_uses)} legacy [[wiki link(s)]]")
print(" use typed markdown links; find/convert them with scripts/list-wiki-links.py")
for w in wiki_uses:
print(f"- {w.from_path}:{w.lineno}: {w.raw}")
def find_invalid_types(graph: Dict[str, object]) -> List[Dict[str, object]]:
return [
e
for e in graph.get("edges", []) # type: ignore[union-attr]
if e.get("type") not in RELATION_TYPES
]
def print_invalid_types(invalid: List[Dict[str, object]]) -> None:
if not invalid:
print("Validation passed: all edge types are in the controlled vocabulary")
return
print(f"Validation failed: {len(invalid)} edge(s) with unknown relationship type")
print(f" allowed: {', '.join(sorted(RELATION_TYPES))}")
for e in invalid:
print(f"- {e.get('from')} -> {e.get('to')}: type={e.get('type')!r}")
def find_invalid_node_types(graph: Dict[str, object]) -> List[Dict[str, object]]:
return [
n
for n in graph.get("nodes", []) # type: ignore[union-attr]
if n.get("type") not in NOTE_TYPES
]
def print_invalid_node_types(invalid: List[Dict[str, object]]) -> None:
if not invalid:
print("Validation passed: all node types are in the controlled vocabulary")
return
print(f"Validation failed: {len(invalid)} note(s) with unknown type")
print(f" allowed: {', '.join(sorted(NOTE_TYPES))}")
for n in invalid:
print(f"- {n.get('id')} ({n.get('path')}): type={n.get('type')!r}")
def _existing_generated_at(output: Path, candidate: GraphBuild) -> str:
"""Keep the old timestamp when the written graph would be unchanged."""
if not output.exists():
return ""
try:
existing = json.loads(output.read_text(encoding="utf-8"))
except (json.JSONDecodeError, OSError, TypeError, ValueError):
return ""
if (
candidate.graph["nodes"] == existing.get("nodes", [])
and candidate.graph["edges"] == existing.get("edges", [])
):
return str(existing.get("generated_at", ""))
return ""
def main() -> None:
args = parse_args()
root = Path(__file__).resolve().parents[1]
root = args.vault_root
output = root / "index" / "graph.json"
should_write = args.write or not args.validate
build = build_graph(root)
if should_write:
stamp = _existing_generated_at(output, build)
if stamp:
build = build_graph(root, existing_generated_at=stamp)
output.parent.mkdir(parents=True, exist_ok=True)
existing_generated_at = ""
if output.exists():
try:
existing = json.loads(output.read_text(encoding="utf-8"))
existing_nodes = existing.get("nodes", [])
existing_edges = existing.get("edges", [])
candidate_graph, unresolved, wiki_uses = generate_graph(root)
if (
candidate_graph["nodes"] == existing_nodes
and candidate_graph["edges"] == existing_edges
):
existing_generated_at = str(existing.get("generated_at", ""))
graph, unresolved, wiki_uses = generate_graph(
root, existing_generated_at=existing_generated_at
)
except (json.JSONDecodeError, OSError, TypeError, ValueError):
graph, unresolved, wiki_uses = generate_graph(root)
else:
graph, unresolved, wiki_uses = generate_graph(root)
if should_write:
output.write_text(json.dumps(graph, indent=2) + "\n", encoding="utf-8")
output.write_text(json.dumps(build.graph, indent=2) + "\n", encoding="utf-8")
print(f"Wrote {output}")
if args.validate:
print_unresolved(unresolved)
invalid = find_invalid_types(graph)
print_invalid_types(invalid)
invalid_nodes = find_invalid_node_types(graph)
print_invalid_node_types(invalid_nodes)
print_wiki_links(wiki_uses)
if unresolved or invalid or invalid_nodes or wiki_uses:
print_unresolved(build.unresolved)
print_invalid_types(build.invalid_edge_types)
print_invalid_node_types(build.invalid_node_types)
print_wiki_links(build.wiki_uses)
if not build.ok:
sys.exit(1)
+43
View File
@@ -0,0 +1,43 @@
#!/usr/bin/env python3
"""Install a vault-local hook that invokes this framework without copying its tools."""
import argparse
import shlex
import subprocess
from pathlib import Path
from vault_root import FRAMEWORK_ROOT, parse_vault_args
def main():
parser = argparse.ArgumentParser(description=__doc__)
root = parse_vault_args(parser).vault_root
repo = subprocess.run(["git", "-C", str(root), "rev-parse", "--show-toplevel"],
capture_output=True, text=True)
if repo.returncode or repo.stdout.strip() != str(root):
parser.error("vault root must be the Git repository root to install hooks")
result = subprocess.run(
["git", "-C", str(root), "rev-parse", "--path-format=absolute", "--git-path", "hooks"],
capture_output=True, text=True, check=True,
)
hook = Path(result.stdout.strip()) / "pre-commit"
marker = "# knowledge-base framework hook"
if hook.exists() or hook.is_symlink():
legacy = hook.is_symlink() and hook.resolve() in {
root / "scripts/pre-commit", FRAMEWORK_ROOT / "scripts/pre-commit",
}
managed = not hook.is_symlink() and marker in hook.read_text()
if not legacy and not managed:
parser.error(f"refusing to replace unmanaged hook: {hook}")
script = FRAMEWORK_ROOT / "scripts/pre-commit"
content = f"#!/bin/sh\n{marker}\nexec bash {shlex.quote(str(script))} {shlex.quote(str(root))}\n"
hook.parent.mkdir(parents=True, exist_ok=True)
if hook.is_symlink():
hook.unlink()
hook.write_text(content)
hook.chmod(0o755)
print(f"Installed {hook} -> {script} (vault: {root})")
if __name__ == "__main__":
main()
+4 -2
View File
@@ -21,6 +21,8 @@ from collections import defaultdict
from pathlib import Path
from typing import Dict, List
from vault_root import parse_vault_args
# Reuse the parser/resolver from generate-graph.py (hyphenated name -> load by path).
_GG_PATH = Path(__file__).resolve().with_name("generate-graph.py")
_spec = importlib.util.spec_from_file_location("generate_graph", _GG_PATH)
@@ -102,9 +104,9 @@ def main() -> None:
default="grouped",
help="output format (default: grouped text)",
)
args = parser.parse_args()
args = parse_vault_args(parser)
root = Path(__file__).resolve().parents[1]
root = args.vault_root
rows = collect(root)
if args.format == "tsv":
print_tsv(rows)
+71 -323
View File
@@ -1,9 +1,12 @@
#!/usr/bin/env python3
"""MCP server for the knowledge-base repository.
"""MCP server for a selected knowledge-base vault, using framework-owned tooling.
This server is the single interface an agent needs to work the vault over stdio
with JSON-RPC 2.0. It exposes graph and note *querying* (graph_summary,
graph_search_nodes, graph_neighbors, read_note, refresh_graph), *writing*
with JSON-RPC 2.0. It provides framework operating instructions (read_instructions),
and exposes graph and note *querying* (graph_summary,
graph_search_nodes, graph_neighbors, read_note — all served by the shared
`kb_framework` package over a graph built in memory from the vault's Markdown
and cached until the next write; refresh_graph rebuilds it), *writing*
(write_note for text, capture_raw for verbatim bytes from a local file or an
http(s) download) and *publishing* (edit_note, git_commit, git_push), so an
agent without shell or filesystem tools can ingest a resource end to end.
@@ -18,6 +21,7 @@ It auto-detects the transport layer:
from __future__ import annotations
import argparse
import hashlib
import json
import shutil
@@ -25,20 +29,35 @@ import subprocess
import sys
import urllib.error
import urllib.request
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Dict, List
from vault_root import FRAMEWORK_ROOT, parse_vault_args # noqa: E402
from kb_framework import graph as kb_graph # noqa: E402
from kb_framework import query_tools # noqa: E402
from kb_framework import vault as kb_vault # noqa: E402
SERVER_NAME = "knowledge-base-mcp"
SERVER_VERSION = "0.1.0"
@dataclass
@dataclass
class AppState:
"""The selected vault plus the graph built from it, cached until a write."""
root: Path
graph_path: Path
graph_cache: Dict[str, Any] | None = None
def graph(self) -> Dict[str, Any]:
if self.graph_cache is None:
self.graph_cache = kb_graph.build_graph(self.root).graph
return self.graph_cache
def invalidate(self) -> None:
self.graph_cache = None
class _StdinReader:
@@ -137,7 +156,6 @@ def _write_message_ndjson(stdout: Any, payload: Dict[str, Any]) -> None:
stdout.buffer.flush()
def _response(message_id: Any, result: Dict[str, Any]) -> Dict[str, Any]:
return {"jsonrpc": "2.0", "id": message_id, "result": result}
@@ -150,47 +168,9 @@ def _error(message_id: Any, code: int, message: str) -> Dict[str, Any]:
}
def _load_graph(state: AppState) -> Dict[str, Any]:
if not state.graph_path.exists():
raise FileNotFoundError(
f"graph not found at {state.graph_path}. Run scripts/generate-graph.py first."
)
return json.loads(state.graph_path.read_text(encoding="utf-8"))
def _nodes_by_id(graph: Dict[str, Any]) -> Dict[str, Dict[str, Any]]:
nodes = graph.get("nodes", [])
return {str(node.get("id")): node for node in nodes if isinstance(node, dict)}
def _match_filters(node: Dict[str, Any], args: Dict[str, Any]) -> bool:
repo = args.get("repo")
note_type = args.get("type")
status = args.get("status")
tag = args.get("tag")
query = str(args.get("query", "")).strip().lower()
if repo and node.get("repo") != repo:
return False
if note_type and node.get("type") != note_type:
return False
if status and node.get("status") != status:
return False
if tag and tag not in (node.get("tags") or []):
return False
if query:
haystacks = [
str(node.get("id", "")).lower(),
str(node.get("title", "")).lower(),
str(node.get("path", "")).lower(),
" ".join(str(t).lower() for t in (node.get("tags") or [])),
]
if not any(query in h for h in haystacks):
return False
return True
_WILDCARD_PATHS = {".", "./", "-A", "--all", "*", "-a", ":/"}
# Reserved for operator-written graph dumps (generate-graph.py --write); never a note.
_GENERATED_GRAPH_PATH = "index/graph.json"
# Ceiling for a single capture_raw call; papers and slide decks fit, a video
# download that would bloat the vault does not.
@@ -199,34 +179,24 @@ _DOWNLOAD_TIMEOUT_SECONDS = 60
# graph_summary defaults: enough most-linked notes to spot the hubs, and enough
# tags to cover the vocabulary an agent actually filters on.
_DEFAULT_TOP_LINKED = 10
_DEFAULT_TOP_TAGS = 50
def _resolve_in_root(state: AppState, raw_path: Any, *, label: str = "path") -> Path:
"""Turn a vault-relative path into an absolute path inside the repo root.
Rejects empty values, absolute paths and anything escaping the root, so a
tool call can never touch a file outside the vault.
"""
value = str(raw_path or "").strip()
if not value:
raise ValueError(f"'{label}' is required")
candidate = Path(value)
if candidate.is_absolute():
raise ValueError(f"'{label}' must be vault-relative, got absolute '{value}'")
root = state.root.resolve()
target = (root / candidate).resolve()
if target != root and not target.is_relative_to(root):
raise ValueError(f"'{label}' resolves outside repository root: '{value}'")
return target
"""Vault-relative path -> absolute path inside the repo root (never outside)."""
return kb_vault.resolve_in_root(state.root, raw_path, label=label)
def _relative_to_root(state: AppState, target: Path) -> str:
return target.relative_to(state.root.resolve()).as_posix()
return kb_vault.relative_to_root(state.root, target)
def _run_git(state: AppState, args: List[str]) -> subprocess.CompletedProcess:
repo = subprocess.run(
["git", "rev-parse", "--show-toplevel"], cwd=state.root,
check=False, capture_output=True, text=True,
)
if repo.returncode != 0 or Path(repo.stdout.strip()).resolve() != state.root.resolve():
raise ValueError("vault root must be the Git repository root for Git operations")
return subprocess.run(
["git", *args],
cwd=state.root,
@@ -249,86 +219,15 @@ def _git_result(payload: Dict[str, Any], exit_code: int) -> Dict[str, Any]:
def _tool_definitions() -> List[Dict[str, Any]]:
return [
{
"name": "graph_summary",
"description": (
"Orientation for a cold start: node and edge counts, the repo "
"scopes, the most-linked notes, and the tag vocabulary as "
"`by_tag` (most frequent first) plus `tag_count`, the number of "
"distinct tags. Read `by_tag` before filtering "
"`graph_search_nodes` by `tag` \u2014 that filter matches a tag "
"verbatim, so a guessed spelling returns nothing."
),
"inputSchema": {
"type": "object",
"properties": {
"topLinked": {
"type": "integer",
"minimum": 1,
"maximum": 50,
"description": "Most-linked notes to return (default 10).",
},
"topTags": {
"type": "integer",
"minimum": 1,
"maximum": 500,
"description": "Tags to return, most frequent first (default 50).",
},
},
"additionalProperties": False,
},
},
{
"name": "graph_search_nodes",
"description": "Search graph nodes by metadata and free text.",
"inputSchema": {
"type": "object",
"properties": {
"repo": {"type": "string"},
"type": {"type": "string"},
"status": {"type": "string"},
"tag": {"type": "string"},
"query": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 200},
},
},
},
{
"name": "graph_neighbors",
"description": (
"Show inbound and outbound links for a node ID. Each neighbor "
"includes the edge `type` (relationship, e.g. supersedes, "
"depends-on), `anchor_text` (the link's display text), and "
"`annotation` (what to expect in that note)."
),
"inputSchema": {
"type": "object",
"required": ["id"],
"properties": {
"id": {"type": "string"},
"limit": {"type": "integer", "minimum": 1, "maximum": 200},
},
},
},
{
"name": "read_note",
"description": (
"Read note content by graph node ID or vault-relative path. "
"Provide exactly one of `id` or `path`. "
"If both or neither are provided, the call returns an error."
),
"inputSchema": {
"type": "object",
"properties": {
"id": {"type": "string", "description": "Graph node ID (e.g. 'AGENTS')."},
"path": {"type": "string", "description": "Vault-relative file path (e.g. 'README.md')."},
},
"additionalProperties": False,
},
},
*query_tools.definitions(),
{
"name": "refresh_graph",
"description": "Regenerate graph.json using scripts/generate-graph.py.",
"description": (
"Rebuild the in-memory graph from the vault's Markdown. With "
"validate=true also run the commit gate (unresolved links, "
"unknown edge/note types, legacy wiki links) and return an error "
"result listing the problems."
),
"inputSchema": {
"type": "object",
"properties": {
@@ -474,175 +373,20 @@ def _text_result(text: str, structured: Dict[str, Any] | None = None) -> Dict[st
return result
def _tool_graph_summary(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
graph = _load_graph(state)
nodes = graph.get("nodes", [])
edges = graph.get("edges", [])
top_n = int(args.get("topLinked", _DEFAULT_TOP_LINKED))
top_tags = int(args.get("topTags", _DEFAULT_TOP_TAGS))
by_repo: Counter[str] = Counter(str(node.get("repo", "unknown")) for node in nodes)
tag_counts: Counter[str] = Counter(
str(tag) for node in nodes for tag in (node.get("tags") or [])
)
# Frequent tags first, alphabetical within a tie, so a truncated list is
# both the most useful slice and stable across calls.
ranked_tags = sorted(tag_counts.items(), key=lambda item: (-item[1], item[0]))
sorted_nodes = sorted(
nodes,
key=lambda n: int(n.get("inbound_count", 0)),
reverse=True,
)[:top_n]
payload = {
"version": graph.get("version"),
"generated_at": graph.get("generated_at"),
"node_count": len(nodes),
"edge_count": len(edges),
"by_repo": dict(by_repo),
"tag_count": len(tag_counts),
"by_tag": [
{"tag": tag, "count": count} for tag, count in ranked_tags[:top_tags]
],
"top_linked": [
{
"id": node.get("id"),
"title": node.get("title"),
"path": node.get("path"),
"inbound_count": node.get("inbound_count", 0),
}
for node in sorted_nodes
],
}
return _text_result(json.dumps(payload, indent=2), structured=payload)
def _tool_graph_search_nodes(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
graph = _load_graph(state)
nodes = graph.get("nodes", [])
limit = int(args.get("limit", 30))
filtered = [node for node in nodes if _match_filters(node, args)]
filtered.sort(key=lambda n: str(n.get("id", "")))
sliced = filtered[:limit]
payload = {
"total_matches": len(filtered),
"returned": len(sliced),
"items": sliced,
}
return _text_result(json.dumps(payload, indent=2), structured=payload)
def _tool_graph_neighbors(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
graph = _load_graph(state)
node_id = str(args["id"])
limit = int(args.get("limit", 50))
nodes = _nodes_by_id(graph)
if node_id not in nodes:
raise ValueError(f"node '{node_id}' not found")
edges = graph.get("edges", [])
outbound_edges = [e for e in edges if e.get("from") == node_id]
inbound_edges = [e for e in edges if e.get("to") == node_id]
def describe(edge_list: List[Dict[str, Any]], direction: str) -> List[Dict[str, Any]]:
# neighbor id is the "other end" relative to node_id
other_key = "to" if direction == "outbound" else "from"
result: List[Dict[str, Any]] = []
for edge in edge_list:
item_id = str(edge.get(other_key))
node = nodes.get(item_id)
entry = {
"id": node.get("id") if node else item_id,
"title": node.get("title") if node else "<missing>",
"path": node.get("path") if node else "",
"direction": direction,
"type": edge.get("type", "related"),
"anchor_text": edge.get("anchor_text", ""),
"annotation": edge.get("annotation", ""),
}
result.append(entry)
return result[:limit]
payload = {
"node": nodes[node_id],
"inbound_count": len(inbound_edges),
"outbound_count": len(outbound_edges),
"inbound": describe(inbound_edges, "inbound"),
"outbound": describe(outbound_edges, "outbound"),
}
return _text_result(json.dumps(payload, indent=2), structured=payload)
def _tool_read_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
has_id = bool(args.get("id"))
has_path = bool(args.get("path"))
if has_id and has_path:
raise ValueError("provide exactly one of 'id' or 'path', not both")
if not has_id and not has_path:
raise ValueError("exactly one of 'id' or 'path' is required")
path_value = args.get("path")
if has_id:
nodes = _nodes_by_id(_load_graph(state))
node = nodes.get(str(args["id"]))
if not node:
raise ValueError(f"node '{args['id']}' not found")
path_value = node.get("path")
target = (state.root / str(path_value)).resolve()
if not target.is_relative_to(state.root.resolve()):
raise ValueError("note path resolves outside repository root")
if not target.exists():
raise FileNotFoundError(f"note not found at {target}")
relative = str(target.relative_to(state.root).as_posix())
data = target.read_bytes()
try:
text = data.decode("utf-8")
except UnicodeDecodeError:
payload = {
"path": relative,
"binary": True,
"bytes": len(data),
"sha256": hashlib.sha256(data).hexdigest(),
"hint": (
f"{target} is not UTF-8 text (e.g. a PDF or image captured under "
"raw/). Open it with a file-reading tool that handles binaries; "
"this server returns text only."
),
}
return _text_result(json.dumps(payload, indent=2), structured=payload)
payload = {
"path": relative,
"binary": False,
"content": text,
}
return _text_result(json.dumps(payload, indent=2), structured=payload)
def _tool_refresh_graph(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
command = [sys.executable, "scripts/generate-graph.py"]
if bool(args.get("validate", False)):
command.append("--validate")
proc = subprocess.run(
command,
cwd=state.root,
check=False,
capture_output=True,
text=True,
)
"""Rebuild the in-memory graph; with validate=true also apply the commit gate."""
build = kb_graph.build_graph(state.root)
state.graph_cache = build.graph
problems = kb_graph.validation_problems(build) if bool(args.get("validate", False)) else []
payload = {
"command": command,
"exit_code": proc.returncode,
"stdout": proc.stdout,
"stderr": proc.stderr,
"vault_root": str(state.root),
"generated_at": build.graph.get("generated_at"),
"node_count": len(build.graph.get("nodes", [])),
"edge_count": len(build.graph.get("edges", [])),
"validated": bool(args.get("validate", False)),
"problems": problems,
}
if proc.returncode != 0:
if problems:
return {
"isError": True,
"content": [{"type": "text", "text": json.dumps(payload, indent=2)}],
@@ -657,7 +401,7 @@ def _tool_write_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
content = str(args["content"])
target = _resolve_in_root(state, args.get("path"))
relative = _relative_to_root(state, target)
if relative == state.graph_path.relative_to(state.root.resolve()).as_posix():
if relative == _GENERATED_GRAPH_PATH:
raise ValueError(
"index/graph.json is generated \u2014 use refresh_graph instead of writing it"
)
@@ -672,6 +416,7 @@ def _tool_write_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_text(content, encoding="utf-8")
state.invalidate()
payload = {
"path": relative,
"created": not existed,
@@ -708,6 +453,7 @@ def _tool_edit_note(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
updated = text.replace(old_string, new_string)
target.write_text(updated, encoding="utf-8")
state.invalidate()
payload = {
"path": relative,
"replacements": occurrences if replace_all else 1,
@@ -784,6 +530,7 @@ def _tool_capture_raw(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(data)
source_label = url
state.invalidate()
payload = {
"path": relative,
@@ -884,14 +631,8 @@ def _tool_git_push(state: AppState, args: Dict[str, Any]) -> Dict[str, Any]:
def _call_tool(state: AppState, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
if name == "graph_summary":
return _tool_graph_summary(state, arguments)
if name == "graph_search_nodes":
return _tool_graph_search_nodes(state, arguments)
if name == "graph_neighbors":
return _tool_graph_neighbors(state, arguments)
if name == "read_note":
return _tool_read_note(state, arguments)
if name in query_tools.READ_ONLY_TOOL_NAMES:
return query_tools.call(name, arguments, graph=state.graph(), root=state.root)
if name == "refresh_graph":
return _tool_refresh_graph(state, arguments)
if name == "write_note":
@@ -908,12 +649,12 @@ def _call_tool(state: AppState, name: str, arguments: Dict[str, Any]) -> Dict[st
def _list_resources(state: AppState) -> Dict[str, Any]:
return {
result = {
"resources": [
{
"uri": "repo://graph",
"name": "Repository Graph",
"description": "Raw index/graph.json",
"description": "The graph built in memory from the vault's Markdown",
"mimeType": "application/json",
},
{
@@ -924,11 +665,17 @@ def _list_resources(state: AppState) -> Dict[str, Any]:
},
]
}
if not (state.root / "README.md").is_file():
result["resources"] = [
resource for resource in result["resources"]
if resource["uri"] != "repo://readme"
]
return result
def _read_resource(state: AppState, uri: str) -> Dict[str, Any]:
if uri == "repo://graph":
text = state.graph_path.read_text(encoding="utf-8")
text = json.dumps(state.graph(), indent=2)
return {
"contents": [
{
@@ -994,8 +741,9 @@ def _handle_request(state: AppState, message: Dict[str, Any]) -> Dict[str, Any]
def main() -> None:
root = Path(__file__).resolve().parents[1]
state = AppState(root=root, graph_path=root / "index" / "graph.json")
args = parse_vault_args(argparse.ArgumentParser(description=__doc__))
root = args.vault_root
state = AppState(root=root)
reader = _StdinReader(sys.stdin.buffer)
try:
+6 -8
View File
@@ -1,11 +1,9 @@
#!/usr/bin/env bash
# Validate the selected vault graph before accepting a commit.
# The graph is derived in memory; nothing is written and nothing must be staged.
set -euo pipefail
echo "[pre-commit] Generating and validating graph..."
make graph
if ! git diff --quiet -- index/graph.json; then
echo "[pre-commit] index/graph.json was modified by generation."
echo "[pre-commit] Run 'make graph' and stage index/graph.json before committing."
exit 1
fi
framework_root="$(cd -- "$(dirname -- "$(readlink -f -- "${BASH_SOURCE[0]}")")/.." && pwd)"
vault_root="${1:-${KNOWLEDGE_BASE_ROOT:-$framework_root}}"
echo "[pre-commit] Validating graph..."
"${PYTHON:-python3}" "$framework_root/scripts/generate-graph.py" --vault-root "$vault_root" --validate
+34
View File
@@ -0,0 +1,34 @@
"""Resolve the selected data directory consistently for framework command-line tools.
Also puts the framework root on `sys.path` so every script can import the
`kb_framework` package from a plain checkout without installing it.
"""
import argparse
import os
import sys
from pathlib import Path
FRAMEWORK_ROOT = Path(__file__).resolve().parents[1]
if str(FRAMEWORK_ROOT) not in sys.path:
sys.path.insert(0, str(FRAMEWORK_ROOT))
def parse_vault_args(parser: argparse.ArgumentParser) -> argparse.Namespace:
"""Add the shared CLI contract and return a validated absolute vault root."""
parser.add_argument(
"--vault-root",
help="data directory (default: KNOWLEDGE_BASE_ROOT, then framework root)",
)
args = parser.parse_args()
raw = args.vault_root if args.vault_root is not None else os.environ.get(
"KNOWLEDGE_BASE_ROOT", str(FRAMEWORK_ROOT)
)
if not raw.strip():
parser.error("vault root must not be empty")
root = Path(raw).expanduser().resolve()
if not root.is_dir():
parser.error(f"vault root must be an existing directory: {root}")
args.vault_root = root
return args
+62 -83
View File
@@ -9,8 +9,23 @@ Capture an external resource into the vault as a reference note, store its
original bytes under `raw/`, decompose it into topics, and link those topics into
the knowledge graph. Follow every step.
Obey `AGENTS.md` and `02_areas/ways-of-working.md` (read them if not already
loaded this session; if `ways-of-working.md` does not exist yet, skip it).
Before ingesting, call `read_instructions` on `knowledge-database` and read both
returned documents: the framework's `AGENTS.md` and `AGENT_QUERY_GUIDE.md`.
The server resolves their location; they are not data-vault nodes and must not
be fetched with `read_note`. If this call fails, report the missing instructions
before changing the vault.
Keep the caller's task instructions in scope. For additional vault policies,
query `graph_search_nodes(type="policy")` and `graph_search_nodes(type="guideline")`,
then `read_note` relevant results and follow their policy links. Do not assume
`02_areas/ways-of-working.md` exists; an empty search means no policies were found,
not that framework or caller instructions cease to apply.
The `knowledge-database` MCP server is the only interface for vault access: use `graph_summary`, `graph_search_nodes`, `graph_neighbors` and
`read_note` to find context; `capture_raw` for source bytes; `write_note` and
`edit_note` for notes; `refresh_graph(validate=true)` to rebuild the in-memory graph and validate.
Use `git_commit` / `git_push` only when the caller explicitly authorizes them.
Skills stay in the framework. Never infer the vault from this skill's location.
The output is a `type: note` reference note at
`03_resources/<resource-name>/index.md`, one `type: note` **topic note** per
@@ -47,10 +62,9 @@ Pick a short kebab-case `<resource-name>` from the title / filename / video
title. Keep it descriptive and unique (e.g. `cyclonedx-sbom-spec`,
`oauth2-rfc6749`).
```bash
ls 03_resources/ # check the slug is not already taken
mkdir -p 03_resources/<resource-name>/raw
```
Check for an existing resource with `graph_search_nodes(query="<resource-name>")`
and `read_note(path="03_resources/<resource-name>/index.md")`. The capture and
write tools create parent directories; no separate folder-creation step is needed.
If the slug already exists, either you are re-ingesting (update in place, tell the
user) or pick a more specific slug.
@@ -59,47 +73,31 @@ user) or pick a more specific slug.
Always preserve the source bytes untouched. Pick the branch that matches step 1.
**Local file** — copy verbatim, keep the original name:
```bash
cp "<source-path>" "03_resources/<resource-name>/raw/"
```
**Local file** — use `capture_raw` with the supplied `source_path` and a
vault-relative `path` such as `03_resources/<resource-name>/raw/<filename>`.
This preserves the original bytes and filename.
**PDF / binary at a URL** — download verbatim:
```bash
curl -fsSL "<url>" -o "03_resources/<resource-name>/raw/<filename>"
```
**PDF / binary at a URL** — use `capture_raw(url="<url>", path=".../raw/<filename>")`.
Check the returned byte count and SHA-256.
**Webpage / link** — record the URL and save the readable content. Use `WebFetch`
to retrieve it (request markdown), then write the result and a `source.url`:
```bash
printf '%s\n' "<url>" > "03_resources/<resource-name>/raw/source.url"
```
Save the fetched body to `raw/page.md` (write the `WebFetch` output to that file).
**Webpage / link** — save the URL using
`write_note(path=".../raw/source.url", content="<url>\n")`. Capture the original
page with `capture_raw(url="<url>", path=".../raw/page.html")`. If the invoking
agent allows `WebFetch`, it may fetch a readable version of that external source;
save the returned body with `write_note(path=".../raw/page.md", content=...)`.
**YouTube video** — pull the transcript into `raw/`. Record the URL first, then
try, in order:
```bash
printf '%s\n' "<url>" > "03_resources/<resource-name>/raw/source.url"
# Preferred: yt-dlp grabs subs + metadata without downloading the video
yt-dlp --skip-download --write-subs --write-auto-subs --sub-lang en \
--convert-subs srt --write-info-json \
-o "03_resources/<resource-name>/raw/%(title)s.%(ext)s" "<url>"
# Fallback if yt-dlp is absent: python transcript API
python3 -c "import youtube_transcript_api" 2>/dev/null && \
python3 -c "from youtube_transcript_api import YouTubeTranscriptApi as A; \
import sys; vid='<VIDEO_ID>'; \
open('03_resources/<resource-name>/raw/transcript.txt','w').write(\
'\n'.join(s['text'] for s in A.get_transcript(vid)))"
```
If neither tool is installed, tell the user which to install (`pip install yt-dlp`
or `pip install youtube-transcript-api`) and stop before fabricating a transcript.
Never invent transcript text.
**YouTube video** — record the URL with `write_note`. If a transcript is supplied,
capture its file with `capture_raw(source_path=..., path=".../raw/transcript.txt")`
or preserve supplied text with `write_note`. These MCP tools do not extract
YouTube transcripts. If none is supplied, report the missing transcript and stop
before writing content notes. Never invent transcript text or run shell extractors.
## 4. Read the captured content
Read what you just stored in `raw/` (the file, `page.md`, the transcript). This —
Use `read_note(path="03_resources/<resource-name>/raw/<filename>")` for captured
text. Path reads access the file directly and work even before the graph picks the file up.
For a captured binary, use the invoking agent’s explicitly permitted inspection
tool on the absolute path returned by `capture_raw`; if unavailable, report it. This —
not the live URL — is the source of truth for the rest of the steps, so the note
matches the bytes we preserved. For long content, read enough to identify every
distinct topic, not just the intro.
@@ -110,24 +108,10 @@ Reason about the **individual topics** inside the resource — the concrete
concepts, standards, formats, tools, or entities it discusses. For each topic,
decide whether the vault already knows about it, and whether it points outward.
For every candidate topic, search the vault for a matching note:
```bash
# free-text over the graph (id / title / path / tags)
jq -r --arg q "<topic>" '.nodes[]
| select((.id|ascii_downcase|contains($q)) or (.title|ascii_downcase|contains($q)))
| "\(.id)\t\(.path)"' index/graph.json
# by tag
jq -r --arg t "<topic-tag>" '.nodes[] | select(.tags[]? == $t) | "\(.id)\t\(.path)"' index/graph.json
# full-text fallback across the PARA buckets
grep -rIl "<topic>" 01_projects 02_areas 03_resources 04_archives
```
Prefer the `knowledge-database` MCP server when available (`graph_search_nodes`,
`graph_neighbors`, `read_note`) — it is the same index. Open promising matches and
confirm they are really the same concept before linking.
For every candidate topic, use `graph_search_nodes(query="<topic>")` and tag
filters from `graph_summary`, then traverse `graph_neighbors` and `read_note`
promising matches. Confirm they cover the same concept before linking.
The query searches metadata, not note bodies; read candidate notes when needed.
Classify each topic:
- **Internal** — the vault has a note for it → link with
@@ -145,7 +129,7 @@ Classify each topic:
## 6. Write a note per topic
Each distinct topic from step 5 becomes its own note at
Use `write_note` to create each distinct topic from step 5 at
`03_resources/<resource-name>/<topic-name>.md` (kebab-case slug). This is what
makes topics first-class, individually linkable graph nodes.
@@ -188,11 +172,11 @@ the *source* needs from the *destination* — "what to expect there".
- Link each topic note back to its resource with
`[<resource title>](resource/<resource-name>){"type":"part-of"}`, and outward to
internal notes (same link form) and external sources (URLs).
- Legacy `[[wiki links]]` are rejected by `make validate-graph` — never write them.
- Legacy `[[wiki links]]` are rejected by `refresh_graph(validate=true)` — never write them.
## 7. Write the resource note (index)
Filename: `03_resources/<resource-name>/index.md`. This is the hub that ties the
Use `write_note` for `03_resources/<resource-name>/index.md`. This hub ties the
raw source to the topic notes.
```md
@@ -230,7 +214,7 @@ last_reviewed: <today, YYYY-MM-DD>
- `ingested` and `last_reviewed` = today (the `currentDate` from session context).
- `type` must stay within the vault taxonomy enum
(`hub|project|area|guideline|policy|feature|roadmap|decision|note|meta|reference`),
which `generate-graph.py --validate` enforces — use `note`;
which `refresh_graph(validate=true)` enforces — use `note`;
the `resource` tag is what marks it.
- Markdown link targets resolve by `id`, rel-path, title, or unique stem — always
use the `id` as the target, e.g. `[Human title](resource/foo/bar)`.
@@ -244,38 +228,33 @@ So the resource is reachable in the graph, not just reaching out:
the most relevant existing notes (the strong internal topics from step 5) — e.g.
under a `## Resources` or `## Related` section, creating it if absent. Add a
`{"type":"…","link-info":"…"}` brace where the relationship is more specific than
`related`. Use individual edits.
`related`. Use `edit_note` for each existing note.
- If several resources accumulate, maintain `03_resources/index.md` as a hub that
lists them, and link the new note from there (create the hub if it does not
exist, with `type: hub`, `tags: [resource, hub]`).
## 9. Regenerate and validate the graph
## 9. Rebuild and validate the graph
```bash
make graph
```
Use `refresh_graph(validate=true)`.
This regenerates `index/graph.json` and validates links and edge relationship
types. If validation reports unresolved links, fix the targets (typos, wrong `id`,
This rebuilds the in-memory graph from the Markdown and validates links, edge
relationship types and note types; the result lists every problem under
`problems` and is an error result while any remain. If validation reports unresolved links, fix the targets (typos, wrong `id`,
or a link to a note that does not exist yet — either create it or downgrade it to
plain text / external URL); if it reports an unknown edge `type`, correct it to a
value in the controlled vocabulary. Re-run until clean. The pre-commit hook rejects
a stale or
invalid graph, so this must pass.
value in the controlled vocabulary. Re-run until clean. The pre-commit hook runs the same validation, so this must
pass. There is no graph file to stage.
## 10. Stage (do not commit unless asked)
Per `ways-of-working.md`, NEVER use `git add -A` or `git add .`. Stage each
touched path individually:
Without explicit publishing authorization, leave the changes uncommitted and
report the touched paths. The MCP interface has no stage-only tool.
```bash
git add 03_resources/<resource-name>/ # note + raw/
git add index/graph.json
git add <any-existing-note-you-backlinked>.md
```
Only commit or open a PR if the user explicitly asks. Changes go via small, clear
pull requests.
When a commit is authorized, use `git_commit(message=..., paths=[...])` with an
explicit list covering the resource folder and every backlinked note.
This tool stages those paths and commits them; do not use wildcard paths.
Use `git_push` only when pushing is explicitly authorized. Follow the caller's
branch and review policy.
## 11. Report
+40 -57
View File
@@ -1,18 +1,28 @@
---
name: knowledge-base-read
description: Read and query the knowledge-base vault — a personal docs-as-code knowledge database (PARA-organized). Use when the user asks what is known about a repo, policy, decision (ADR), area, or project; wants context before working in a repo; or asks to find/navigate notes by tag, type, repo, or link. Resolves the question against the Markdown notes and the generated graph index.
description: Read and query the knowledge-base vault — a personal docs-as-code knowledge database (PARA-organized). Use when the user asks what is known about a repo, policy, decision (ADR), area, or project; wants context before working in a repo; or asks to find/navigate notes by tag, type, repo, or link. Resolves the question against the Markdown notes and the graph the MCP server builds from them in memory.
---
# Read the Knowledge Base
This vault is a personal docs-as-code knowledge database,
organized with the [PARA method](https://fortelabs.com/blog/para/).
Markdown is the source of truth; `index/graph.json` is a generated navigation
index. Use this skill to answer questions FROM the vault and to surface the
Markdown is the source of truth; the graph is a navigation index the MCP
server builds from it in memory (nothing on disk). Use this skill to answer questions FROM the vault and to surface the
exact notes that back the answer.
Obey `AGENTS.md` (read them if not loaded this
session). This skill is read-only — do not modify notes here.
Call `read_instructions` on `knowledge-database` to read the framework's
`AGENTS.md` and `AGENT_QUERY_GUIDE.md` before querying the vault. These documents
are not data-vault nodes. If unavailable, report the missing instructions.
This skill is read-only — do not modify notes here.
## Vault access
The `knowledge-database` MCP server is the only interface for vault access. Use `graph_summary` to orient, `graph_search_nodes` to find notes,
`graph_neighbors` to traverse links, and `read_note` to read the source.
Use `refresh_graph` when the graph looks stale. Skills and operational
documentation belong to the framework; never infer the data vault from this
skill's file location.
## Layout (PARA)
@@ -24,85 +34,58 @@ session). This skill is read-only — do not modify notes here.
03_resources/repos/<repo>/ per-repo info.md (+ decisions/)
03_resources/repos/repo-landscape.md index of documented repos
04_archives/ inactive items
index/graph.json generated graph (nodes + edges)
README.md vault hub; AGENT_QUERY_GUIDE.md graph query reference
```
Every note carries taxonomy frontmatter: `id`, `title`, `type`
(`hub|project|area|guideline|policy|feature|roadmap|decision|note|meta|reference`, enforced by
`generate-graph.py --validate`), `status`
`refresh_graph(validate=true)`), `status`
(`active|draft|deprecated`), `tags`, `last_reviewed`, optional `repo`. Notes are
linked with typed markdown links — `[text](note-id){"type":"…"}` — that resolve
by `id`. Legacy `[[wiki links]]` are rejected by `make validate-graph`.
by `id`. Legacy `[[wiki links]]` are rejected by `refresh_graph(validate=true)`.
## 1. Orient
Start broad, then narrow:
- `README.md` — vault hub and entry points.
- `03_resources/repos/repo-landscape.md` — index of every documented repo, grouped.
- `02_areas/ways-of-working.md` — global engineering policy.
- `03_resources/meta/taxonomy.md` — what the frontmatter fields and PARA buckets mean.
For a repo question, read `03_resources/repos/<repo>/info.md` directly
(`ls 03_resources/repos/` to confirm the repo is documented; if absent, say so —
it is not yet in the vault).
Start broad, then narrow. Begin with `graph_summary`, then find hubs using
`graph_search_nodes(type="hub")` and read relevant ones with `read_note`.
Discover repo scopes and tags from the graph rather than assuming a particular
folder layout. An external data vault need not contain a root README or
framework operational documents.
## 2. Query the graph
`index/graph.json` is for navigation by tag / type / repo / links. Use `jq`.
Full schema and more examples are in `AGENT_QUERY_GUIDE.md`.
Use the MCP tools directly:
```bash
# all documented repos
jq -r '.nodes[].repo' index/graph.json | sort -u
| Need | Tool |
|------|------|
| Repo scopes, tag spellings, most-linked notes | `graph_summary` |
| Notes in a repo | `graph_search_nodes(repo="<repo>")` |
| Notes of a type | `graph_search_nodes(type="policy")` |
| Notes with a tag | `graph_search_nodes(tag="<tag>")` |
| Search ids, titles, paths and tags | `graph_search_nodes(query="<term>")` |
| Incoming/outgoing links, relationship types and annotations | `graph_neighbors(id="<note-id>")` |
# every note in one repo
jq -r '.nodes[] | select(.repo=="tenant-controller") | "\(.id)\t\(.path)"' index/graph.json
# all notes of a type (e.g. policies, decisions, projects)
jq -r '.nodes[] | select(.type=="policy") | "\(.id)\t\(.path)"' index/graph.json
# notes carrying a tag
jq -r '.nodes[] | select(.tags[]? == "kcp") | "\(.id)\t\(.path)"' index/graph.json
# most-linked (central) notes
jq -r '.nodes | sort_by(-(.inbound_count // 0))[] | "\(.inbound_count // 0)\t\(.id)"' index/graph.json | head
# outgoing links from a note, with relationship type + annotation
NOTE_ID="repo/controller-util/info"
jq -r --arg id "$NOTE_ID" '.edges[] | select(.from==$id) | "\(.type)\t\(.to)\t\(.annotation)"' index/graph.json
# incoming links to a note (who references it, and how)
jq -r --arg id "$NOTE_ID" '.edges[] | select(.to==$id) | "\(.type)\t\(.from)\t\(.annotation)"' index/graph.json
# edges of a specific relationship type (e.g. supersedes, depends-on)
jq -r '.edges[] | select(.type=="supersedes") | "\(.from) -> \(.to)"' index/graph.json
```
Full-text fallback when you do not know the tag/type/id:
```bash
grep -rIl "<term>" 01_projects 02_areas 03_resources 04_archives
```
`query` does not search note bodies. Read candidate notes with `read_note` and
follow their neighbors when metadata search is insufficient. Report any search
coverage limits; do not use filesystem or shell searches.
## 3. Read the source notes
The graph gives ids and paths; the ANSWER lives in the Markdown. Open the
matched `.path` files and read them. Follow the typed markdown links
The graph gives ids and paths; the ANSWER lives in the Markdown. Read matched
notes with `read_note(id="<note-id>")` or `read_note(path="<vault-relative-path>")`. Follow the typed markdown links
`[text](id){"type":"…"}`, resolving each to its node `path` to pull in related
policies, decisions, and repo info. Use each edge's `type` and `annotation` to
judge which links are worth following; links are the intended navigation, so
traverse them rather than guessing.
If the graph looks stale or a path is missing, you may regenerate read-only:
`python3 scripts/generate-graph.py` (does not touch Markdown). Do not commit.
If the graph looks stale or a path is missing, use `refresh_graph`. This rebuilds
the in-memory graph; it does not touch Markdown. Nothing to commit.
## 4. Answer with citations
- Answer the question directly from what the notes say.
- Cite the backing notes by vault-relative `path` (e.g.
`03_resources/repos/gittea/info.md`) and/or `id` so the user can verify.
`03_resources/<resource>/index.md`) and/or `id` so the user can verify.
- Distinguish vault facts from inference. If the vault does not cover it, say so
plainly — do not invent. Note `status: draft`/`deprecated` when relevant.
- For "what links to / relates to X" questions, report the inbound/outbound
+105
View File
@@ -0,0 +1,105 @@
{
"version": 1,
"nodes": [
{
"id": "README",
"title": "README",
"path": "README.md",
"repo": "global",
"type": "note",
"status": "active",
"tags": [],
"last_reviewed": "",
"outbound_links": [],
"inbound_count": 0
},
{
"id": "area-ops",
"title": "Area Ops",
"path": "02_areas/ops.md",
"repo": "global",
"type": "area",
"status": "active",
"tags": [
"ops"
],
"last_reviewed": "",
"outbound_links": [
"project-alpha"
],
"inbound_count": 2
},
{
"id": "project-alpha",
"title": "Project Alpha",
"path": "01_projects/alpha.md",
"repo": "alpha",
"type": "project",
"status": "active",
"tags": [
"alpha",
"demo"
],
"last_reviewed": "2026-01-01",
"outbound_links": [
"area-ops",
"resource-demo"
],
"inbound_count": 1
},
{
"id": "resource-demo",
"title": "Demo Resource",
"path": "03_resources/demo/index.md",
"repo": "global",
"type": "reference",
"status": "active",
"tags": [
"demo"
],
"last_reviewed": "",
"outbound_links": [
"area-ops"
],
"inbound_count": 1
}
],
"edges": [
{
"from": "area-ops",
"to": "project-alpha",
"kind": "markdown_link",
"type": "part-of",
"anchor_text": "Project Alpha",
"annotation": "",
"weight": 1.0
},
{
"from": "project-alpha",
"to": "area-ops",
"kind": "markdown_link",
"type": "depends-on",
"anchor_text": "Area Ops",
"annotation": "operations area",
"weight": 1.0
},
{
"from": "project-alpha",
"to": "resource-demo",
"kind": "markdown_link",
"type": "related",
"anchor_text": "the demo resource",
"annotation": "resource landing page.",
"weight": 1.0
},
{
"from": "resource-demo",
"to": "area-ops",
"kind": "markdown_link",
"type": "related",
"anchor_text": "Area Ops",
"annotation": "",
"weight": 1.0
}
]
}
+259
View File
@@ -0,0 +1,259 @@
{
"graph_summary": {
"tool": "graph_summary",
"args": {
"topLinked": 3,
"topTags": 2
},
"structuredContent": {
"version": 1,
"node_count": 4,
"edge_count": 4,
"by_repo": {
"global": 3,
"alpha": 1
},
"tag_count": 3,
"by_tag": [
{
"tag": "demo",
"count": 2
},
{
"tag": "alpha",
"count": 1
}
],
"top_linked": [
{
"id": "area-ops",
"title": "Area Ops",
"path": "02_areas/ops.md",
"inbound_count": 2
},
{
"id": "project-alpha",
"title": "Project Alpha",
"path": "01_projects/alpha.md",
"inbound_count": 1
},
{
"id": "resource-demo",
"title": "Demo Resource",
"path": "03_resources/demo/index.md",
"inbound_count": 1
}
]
}
},
"graph_search_nodes": {
"tool": "graph_search_nodes",
"args": {
"query": "alpha",
"limit": 5
},
"structuredContent": {
"total_matches": 1,
"returned": 1,
"items": [
{
"id": "project-alpha",
"title": "Project Alpha",
"path": "01_projects/alpha.md",
"repo": "alpha",
"type": "project",
"status": "active",
"tags": [
"alpha",
"demo"
],
"last_reviewed": "2026-01-01",
"outbound_links": [
"area-ops",
"resource-demo"
],
"inbound_count": 1
}
]
}
},
"graph_neighbors": {
"tool": "graph_neighbors",
"args": {
"id": "area-ops"
},
"structuredContent": {
"node": {
"id": "area-ops",
"title": "Area Ops",
"path": "02_areas/ops.md",
"repo": "global",
"type": "area",
"status": "active",
"tags": [
"ops"
],
"last_reviewed": "",
"outbound_links": [
"project-alpha"
],
"inbound_count": 2
},
"inbound_count": 2,
"outbound_count": 1,
"inbound": [
{
"id": "project-alpha",
"title": "Project Alpha",
"path": "01_projects/alpha.md",
"direction": "inbound",
"type": "depends-on",
"anchor_text": "Area Ops",
"annotation": "operations area"
},
{
"id": "resource-demo",
"title": "Demo Resource",
"path": "03_resources/demo/index.md",
"direction": "inbound",
"type": "related",
"anchor_text": "Area Ops",
"annotation": ""
}
],
"outbound": [
{
"id": "project-alpha",
"title": "Project Alpha",
"path": "01_projects/alpha.md",
"direction": "outbound",
"type": "part-of",
"anchor_text": "Project Alpha",
"annotation": ""
}
]
}
},
"read_note_by_id": {
"tool": "read_note",
"args": {
"id": "resource-demo"
},
"structuredContent": {
"path": "03_resources/demo/index.md",
"binary": false,
"content": "---\nid: resource-demo\ntitle: Demo Resource\ntype: reference\ntags: [demo]\n---\n\n# Demo Resource\n\nRaw original kept in `raw/`. Related: [Area Ops](02_areas/ops.md).\n"
}
},
"read_note_by_path": {
"tool": "read_note",
"args": {
"path": "02_areas/ops.md"
},
"structuredContent": {
"path": "02_areas/ops.md",
"binary": false,
"content": "---\nid: area-ops\ntitle: Area Ops\ntype: area\ntags: [ops]\n---\n\n# Area Ops\n\nParent hub: [Project Alpha](project-alpha){\"type\":\"part-of\"}.\n"
}
},
"tool_definitions": [
{
"name": "read_instructions",
"description": "Read framework AGENTS.md and AGENT_QUERY_GUIDE.md before working with the vault. These operating documents live outside an external vault and do not require a generated graph.",
"inputSchema": {
"type": "object",
"properties": {},
"additionalProperties": false
}
},
{
"name": "graph_summary",
"description": "Orientation for a cold start: node and edge counts, the repo scopes, the most-linked notes, and the tag vocabulary as `by_tag` (most frequent first) plus `tag_count`, the number of distinct tags. Read `by_tag` before filtering `graph_search_nodes` by `tag` \u2014 that filter matches a tag verbatim, so a guessed spelling returns nothing.",
"inputSchema": {
"type": "object",
"properties": {
"topLinked": {
"type": "integer",
"minimum": 1,
"maximum": 50,
"description": "Most-linked notes to return (default 10)."
},
"topTags": {
"type": "integer",
"minimum": 1,
"maximum": 500,
"description": "Tags to return, most frequent first (default 50)."
}
},
"additionalProperties": false
}
},
{
"name": "graph_search_nodes",
"description": "Search graph nodes by metadata and free text.",
"inputSchema": {
"type": "object",
"properties": {
"repo": {
"type": "string"
},
"type": {
"type": "string"
},
"status": {
"type": "string"
},
"tag": {
"type": "string"
},
"query": {
"type": "string"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 200
}
}
}
},
{
"name": "graph_neighbors",
"description": "Show inbound and outbound links for a node ID. Each neighbor includes the edge `type` (relationship, e.g. supersedes, depends-on), `anchor_text` (the link's display text), and `annotation` (what to expect in that note).",
"inputSchema": {
"type": "object",
"required": [
"id"
],
"properties": {
"id": {
"type": "string"
},
"limit": {
"type": "integer",
"minimum": 1,
"maximum": 200
}
}
}
},
{
"name": "read_note",
"description": "Read note content by graph node ID or vault-relative path. Provide exactly one of `id` or `path`. If both or neither are provided, the call returns an error.",
"inputSchema": {
"type": "object",
"properties": {
"id": {
"type": "string",
"description": "Graph node ID (e.g. 'AGENTS')."
},
"path": {
"type": "string",
"description": "Vault-relative file path (e.g. 'README.md')."
}
},
"additionalProperties": false
}
}
]
}
+15
View File
@@ -0,0 +1,15 @@
---
id: project-alpha
title: Project Alpha
type: project
status: active
tags: [alpha, demo]
repo: alpha
last_reviewed: 2026-01-01
---
# Project Alpha
Depends on [Area Ops](area-ops){"type":"depends-on","link-info":"operations area"}.
See [the demo resource](resource-demo) — resource landing page.
External: [MCP spec](https://modelcontextprotocol.io/).
+10
View File
@@ -0,0 +1,10 @@
---
id: area-ops
title: Area Ops
type: area
tags: [ops]
---
# Area Ops
Parent hub: [Project Alpha](project-alpha){"type":"part-of"}.
+10
View File
@@ -0,0 +1,10 @@
---
id: resource-demo
title: Demo Resource
type: reference
tags: [demo]
---
# Demo Resource
Raw original kept in `raw/`. Related: [Area Ops](02_areas/ops.md).
@@ -0,0 +1,2 @@
not a note
[broken](../missing.md)
Binary file not shown.
+1
View File
@@ -0,0 +1 @@
# README
+229
View File
@@ -0,0 +1,229 @@
"""Exercise framework commands against separate vaults to prevent cross-repo writes."""
import json
import os
import shutil
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
FRAMEWORK = Path(__file__).resolve().parents[1]
class ExternalVaultTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
self.base = Path(self.tmp.name)
self.framework = self.base / "framework space"
shutil.copytree(FRAMEWORK / "scripts", self.framework / "scripts")
shutil.copytree(FRAMEWORK / "kb_framework", self.framework / "kb_framework",
ignore=shutil.ignore_patterns("__pycache__"))
shutil.copy2(FRAMEWORK / "Makefile", self.framework / "Makefile")
self.vault = self.base / "vault space"
self.other = self.base / "other"
self.env = {k: v for k, v in os.environ.items() if k != "KNOWLEDGE_BASE_ROOT"}
for root, name in [(self.framework, "framework"), (self.vault, "vault"), (self.other, "other")]:
(root / "03_resources").mkdir(parents=True, exist_ok=True)
(root / "03_resources" / "index.md").write_text(
f"---\nid: {name}\ntitle: {name}\ntype: hub\n---\n# {name}\n"
)
self.git(root, "init", "-q", "-b", "main")
self.git(root, "config", "user.email", "test@example.com")
self.git(root, "config", "user.name", "Test")
self.git(root, "config", "commit.gpgsign", "false")
self.git(root, "add", "03_resources/index.md")
self.git(root, "commit", "-qm", "init")
def git(self, root, *args):
return subprocess.run(["git", *args], cwd=root, env=self.env,
capture_output=True, text=True, check=True).stdout.strip()
def run_script(self, name, *args, input="", env=None):
return subprocess.run([sys.executable, str(self.framework / "scripts" / name), *map(str, args)],
cwd=self.other, env=env or self.env, input=input,
capture_output=True, text=True, timeout=15)
def rpc(self, calls, *args, env=None):
requests = [{"jsonrpc": "2.0", "id": i, "method": method, "params": params}
for i, (method, params) in enumerate(calls)]
proc = self.run_script("mcp-server.py", *args, env=env,
input="".join(json.dumps(r) + "\n" for r in requests))
self.assertEqual(proc.returncode, 0, proc.stderr)
responses = [json.loads(line) for line in proc.stdout.splitlines()]
self.assertEqual(len(responses), len(calls), proc.stdout)
for response in responses:
self.assertNotIn("error", response, response)
return [r["result"] for r in responses]
def tool(self, name, arguments=None, root=None):
return self.rpc([("tools/call", {"name": name, "arguments": arguments or {}})],
"--vault-root", root or self.vault)[0]
def test_generator_writes_only_external_vault(self):
proc = self.run_script("generate-graph.py", "--vault-root", self.vault, "--write", "--validate")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
graph = json.loads((self.vault / "index/graph.json").read_text())
self.assertEqual([n["id"] for n in graph["nodes"]], ["vault"])
self.assertFalse((self.framework / "index/graph.json").exists())
self.assertFalse((self.other / "index/graph.json").exists())
def test_instructions_are_discoverable_and_come_from_framework_without_graph(self):
expected = {}
for name in ["AGENTS.md", "AGENT_QUERY_GUIDE.md"]:
expected[name] = f"# Framework {name}\nFramework-owned instructions.\n"
(self.framework / name).write_text(expected[name])
(self.vault / name).write_text("# Unrelated vault file\n")
listed = self.rpc([("tools/list", {})], "--vault-root", self.vault)[0]
self.assertIn("read_instructions", {t["name"] for t in listed["tools"]})
result = self.tool("read_instructions")
self.assertFalse(result.get("isError"), result)
documents = result["structuredContent"]["documents"]
self.assertEqual({d["path"]: d["content"] for d in documents}, expected)
self.assertEqual(result["structuredContent"]["framework_root"], str(self.framework))
self.assertFalse((self.vault / "index/graph.json").exists())
def test_missing_framework_instructions_fail_explicitly_without_vault_fallback(self):
(self.framework / "AGENTS.md").write_text("Read AGENT_QUERY_GUIDE.md\n")
(self.vault / "AGENT_QUERY_GUIDE.md").write_text("Wrong instructions\n")
result = self.tool("read_instructions")
self.assertTrue(result.get("isError"), result)
message = result["content"][0]["text"]
self.assertIn(str(self.framework / "AGENT_QUERY_GUIDE.md"), message)
def test_wiki_inventory_uses_external_vault(self):
for root, target in [(self.vault, "vault-target"), (self.framework, "framework-target")]:
(root / "03_resources/wiki.md").write_text(f"[[{target}]]\n")
proc = self.run_script("list-wiki-links.py", "--vault-root", self.vault)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn("vault-target", proc.stdout)
self.assertNotIn("framework-target", proc.stdout)
def test_mcp_reads_writes_and_commits_only_vault(self):
framework_head = self.git(self.framework, "rev-parse", "HEAD")
result = self.tool("write_note", {"path": "03_resources/new.md", "content": "# New\n"})
self.assertFalse(result.get("isError"), result)
read = self.tool("read_note", {"path": "03_resources/new.md"})
self.assertEqual(read["structuredContent"]["content"], "# New\n")
result = self.tool("git_commit", {"message": "add test note", "paths": ["03_resources/new.md"]})
self.assertFalse(result.get("isError"), result)
self.assertEqual(self.git(self.vault, "log", "-1", "--format=%s"), "add test note")
self.assertEqual(self.git(self.framework, "rev-parse", "HEAD"), framework_head)
self.assertFalse((self.framework / "03_resources/new.md").exists())
for tool, args in [("read_note", {"path": "../framework space/03_resources/index.md"}),
("write_note", {"path": "../escape.md", "content": "escape"})]:
self.assertTrue(self.tool(tool, args).get("isError"))
def test_refresh_validates_in_memory_and_never_runs_vault_tools(self):
(self.vault / "scripts").mkdir()
(self.vault / "scripts/generate-graph.py").write_text("raise RuntimeError('vault tool used')\n")
result = self.tool("refresh_graph", {"validate": True})
self.assertFalse(result.get("isError"), result)
self.assertEqual(result["structuredContent"]["vault_root"], str(self.vault))
self.assertEqual(result["structuredContent"]["node_count"], 1)
self.assertFalse((self.vault / "index/graph.json").exists())
summary = self.tool("graph_summary")
self.assertFalse(summary.get("isError"), summary)
def test_invalid_roots_fail_clearly_for_all_commands(self):
file = self.base / "file"
file.write_text("not a directory")
for script in ["generate-graph.py", "list-wiki-links.py", "mcp-server.py"]:
for root in [self.base / "missing", file, ""]:
with self.subTest(script=script, root=root):
proc = self.run_script(script, "--vault-root", root)
self.assertNotEqual(proc.returncode, 0)
self.assertIn("vault root", proc.stderr.lower())
self.assertNotIn("Traceback", proc.stderr)
def test_root_precedence_and_standalone_for_all_commands(self):
for root, name in [(self.framework, "framework"), (self.vault, "vault"), (self.other, "other")]:
(root / "03_resources/wiki.md").write_text(f"[[{name}-target]]\n")
for args, env, root, name in [([], self.env, self.framework, "framework"),
([], dict(self.env, KNOWLEDGE_BASE_ROOT=str(self.other)), self.other, "other"),
(["--vault-root", str(self.vault)], dict(self.env, KNOWLEDGE_BASE_ROOT=str(self.other)), self.vault, "vault")]:
with self.subTest(name=name):
proc = self.run_script("generate-graph.py", *args, env=env)
self.assertEqual(proc.returncode, 0, proc.stderr)
graph = json.loads((root / "index/graph.json").read_text())
self.assertIn(name, [n["id"] for n in graph["nodes"]])
proc = self.run_script("list-wiki-links.py", *args, env=env)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn(f"{name}-target", proc.stdout)
result = self.rpc([("tools/call", {"name": "read_note", "arguments": {"path": "03_resources/index.md"}})], *args, env=env)[0]
self.assertIn(f"# {name}", result["structuredContent"]["content"])
def test_mcp_protocol_and_optional_readme_resource(self):
self.tool("refresh_graph")
results = self.rpc([("initialize", {}), ("tools/list", {}), ("resources/list", {})], "--vault-root", self.vault)
self.assertEqual(results[0]["serverInfo"]["name"], "knowledge-base-mcp")
self.assertIn("graph_summary", {t["name"] for t in results[1]["tools"]})
self.assertNotIn("repo://readme", {r["uri"] for r in results[2]["resources"]})
(self.vault / "README.md").write_text("# Optional hub\n")
result = self.rpc([("resources/list", {})], "--vault-root", self.vault)[0]
self.assertIn("repo://readme", {r["uri"] for r in result["resources"]})
def test_git_refuses_parent_repository_for_nested_non_repo_vault(self):
nested = self.framework / "nested"
nested.mkdir()
(nested / "note.md").write_text("# note\n")
before = self.git(self.framework, "rev-parse", "HEAD")
result = self.tool("git_commit", {"message": "must fail", "paths": ["note.md"]}, root=nested)
self.assertTrue(result.get("isError"), result)
self.assertEqual(self.git(self.framework, "rev-parse", "HEAD"), before)
def test_make_and_installed_hook_use_selected_vault(self):
def make(*args):
return subprocess.run(["make", "-C", str(self.framework), f"VAULT_ROOT={self.vault}", *args],
env=self.env, capture_output=True, text=True)
proc = make("validate-graph", "install-hooks")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
hook = self.vault / ".git/hooks/pre-commit"
self.assertTrue(hook.exists())
proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True)
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
(self.vault / "03_resources/new.md").write_text("# new\n[gone](nowhere)\n")
proc = subprocess.run([str(hook)], cwd=self.other, env=self.env, capture_output=True, text=True)
self.assertNotEqual(proc.returncode, 0)
self.assertIn("nowhere", proc.stdout + proc.stderr)
self.assertFalse((self.vault / "index/graph.json").exists())
self.assertFalse((self.framework / "index/graph.json").exists())
def test_installer_preserves_unmanaged_hooks_and_reinstalls_managed_hook(self):
hook = self.vault / ".git/hooks/pre-commit"
custom = "#!/bin/sh\n# user's hook\nexit 0\n"
hook.write_text(custom)
proc = self.run_script("install-hooks.py", "--vault-root", self.vault)
self.assertNotEqual(proc.returncode, 0)
self.assertIn("unmanaged hook", proc.stderr)
self.assertEqual(hook.read_text(), custom)
hook.unlink()
(self.vault / "scripts").mkdir()
target = self.vault / "scripts/pre-commit"
target.write_text(custom)
hook.symlink_to("../../scripts/pre-commit")
for _ in range(2):
proc = self.run_script("install-hooks.py", "--vault-root", self.vault)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertFalse(hook.is_symlink())
self.assertEqual(target.read_text(), custom)
def test_installer_honors_worktree_and_custom_hooks_directory(self):
worktree = self.base / "worktree"
self.git(self.vault, "worktree", "add", "-qb", "worktree", str(worktree))
proc = self.run_script("install-hooks.py", "--vault-root", worktree)
self.assertEqual(proc.returncode, 0, proc.stderr)
hook = Path(self.git(worktree, "rev-parse", "--path-format=absolute", "--git-path", "hooks/pre-commit"))
self.assertIn(str(worktree), hook.read_text())
custom = self.base / "custom hooks"
self.git(self.vault, "config", "core.hooksPath", str(custom))
proc = self.run_script("install-hooks.py", "--vault-root", self.vault)
self.assertEqual(proc.returncode, 0, proc.stderr)
self.assertIn(str(self.vault), (custom / "pre-commit").read_text())
if __name__ == "__main__":
unittest.main()
+2 -7
View File
@@ -63,12 +63,7 @@ class TestValidateExitCode(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name).resolve() / "vault"
(self.root / "scripts").mkdir(parents=True)
(self.root / "03_resources").mkdir()
(self.root / "scripts" / "generate-graph.py").write_text(
(SCRIPTS / "generate-graph.py").read_text(encoding="utf-8"),
encoding="utf-8",
)
(self.root / "03_resources").mkdir(parents=True)
def tearDown(self) -> None:
self._tmp.cleanup()
@@ -81,7 +76,7 @@ class TestValidateExitCode(unittest.TestCase):
def validate(self) -> subprocess.CompletedProcess:
return subprocess.run(
[sys.executable, "scripts/generate-graph.py", "--validate"],
[sys.executable, str(SCRIPTS / "generate-graph.py"), "--vault-root", str(self.root), "--validate"],
cwd=self.root,
capture_output=True,
text=True,
+124
View File
@@ -0,0 +1,124 @@
#!/usr/bin/env python3
"""Tests for the graph-in-memory contract of the stdio server and the hook.
`index/graph.json` is no longer a committed artifact: the MCP server builds
the graph lazily from Markdown, invalidates it on writes, and `refresh_graph`
rebuilds (and optionally validates) in memory. The pre-commit hook only
validates. These tests pin that nothing in the default flow writes the file.
"""
from __future__ import annotations
import importlib
import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
FRAMEWORK = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(FRAMEWORK / "scripts"))
mcp = importlib.import_module("mcp-server")
def write_note(root: Path, rel: str, note_id: str, body: str = "") -> None:
path = root / rel
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(f"---\nid: {note_id}\ntitle: {note_id}\ntype: note\n---\n# {note_id}\n{body}\n", encoding="utf-8")
class InMemoryVault(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name).resolve() / "vault"
self.root.mkdir()
write_note(self.root, "03_resources/a.md", "a", "[b](b)")
write_note(self.root, "03_resources/b.md", "b")
self.state = mcp.AppState(root=self.root)
def tearDown(self) -> None:
self._tmp.cleanup()
def call(self, name: str, **args) -> dict:
return mcp._call_tool(self.state, name, args)
class TestStdioServerBuildsGraphLazily(InMemoryVault):
def test_summary_without_graph_file(self) -> None:
payload = self.call("graph_summary")["structuredContent"]
self.assertEqual(payload["node_count"], 2)
self.assertEqual(payload["edge_count"], 1)
self.assertFalse((self.root / "index").exists())
def test_write_note_invalidates_cache(self) -> None:
self.call("graph_summary")
self.call("write_note", path="03_resources/c.md", content="---\nid: c\n---\n# c\n")
self.assertEqual(self.call("graph_summary")["structuredContent"]["node_count"], 3)
def test_edit_note_invalidates_cache(self) -> None:
self.call("graph_summary")
self.call("edit_note", path="03_resources/b.md", old_string="# b", new_string="# b\n[a](a)")
self.assertEqual(self.call("graph_summary")["structuredContent"]["edge_count"], 2)
def test_refresh_rebuilds_and_validates_in_memory(self) -> None:
self.call("graph_summary")
write_note(self.root, "03_resources/d.md", "d", "[gone](nowhere)")
ok = self.call("refresh_graph")
self.assertFalse(ok.get("isError"), ok)
self.assertEqual(ok["structuredContent"]["node_count"], 3)
bad = self.call("refresh_graph", validate=True)
self.assertTrue(bad.get("isError"), bad)
self.assertIn("nowhere", json.dumps(bad["structuredContent"]))
self.assertFalse((self.root / "index").exists())
def test_graph_resource_is_served_from_memory(self) -> None:
result = mcp._read_resource(self.state, "repo://graph")
graph = json.loads(result["contents"][0]["text"])
self.assertEqual(len(graph["nodes"]), 2)
self.assertFalse((self.root / "index").exists())
def test_read_note_by_id_uses_in_memory_graph(self) -> None:
payload = self.call("read_note", id="b")["structuredContent"]
self.assertEqual(payload["path"], "03_resources/b.md")
class TestGeneratorAndHookValidateOnly(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name).resolve() / "vault"
self.root.mkdir()
write_note(self.root, "03_resources/a.md", "a")
self.env = {k: v for k, v in os.environ.items() if k != "KNOWLEDGE_BASE_ROOT"}
def tearDown(self) -> None:
self._tmp.cleanup()
def run_cmd(self, *cmd: str) -> subprocess.CompletedProcess:
return subprocess.run(list(cmd), cwd=self.root, env=self.env, capture_output=True, text=True)
def test_validate_alone_writes_nothing(self) -> None:
proc = self.run_cmd(sys.executable, str(FRAMEWORK / "scripts/generate-graph.py"), "--vault-root", str(self.root), "--validate")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertFalse((self.root / "index").exists())
def test_write_still_produces_a_file_for_operators(self) -> None:
proc = self.run_cmd(sys.executable, str(FRAMEWORK / "scripts/generate-graph.py"), "--vault-root", str(self.root), "--write")
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertTrue((self.root / "index/graph.json").exists())
def test_hook_validates_without_requiring_a_staged_graph(self) -> None:
hook = FRAMEWORK / "scripts/pre-commit"
proc = self.run_cmd("bash", str(hook), str(self.root))
self.assertEqual(proc.returncode, 0, proc.stdout + proc.stderr)
self.assertFalse((self.root / "index").exists())
write_note(self.root, "03_resources/b.md", "b", "[x](missing)")
proc = self.run_cmd("bash", str(hook), str(self.root))
self.assertNotEqual(proc.returncode, 0)
self.assertIn("missing", proc.stdout + proc.stderr)
self.assertFalse((self.root / "index").exists())
if __name__ == "__main__":
unittest.main()
+162
View File
@@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""Tests for the importable `kb_framework` package.
The package is the contract both the stdio MCP server and the hosted connector
consume: build the graph in memory from a vault's Markdown, answer the query
tools over that graph, and read framework instructions. These tests pin the
graph shape against a golden fixture produced by the file-writing generator,
and the tool payloads against goldens captured from the stdio server, so the
two front ends cannot drift apart.
"""
from __future__ import annotations
import json
import sys
import unittest
from pathlib import Path
FRAMEWORK = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(FRAMEWORK))
FIXTURES = FRAMEWORK / "tests" / "fixtures"
VAULT = FIXTURES / "vault"
from kb_framework import graph as kbg # noqa: E402
from kb_framework import query as kbq # noqa: E402
from kb_framework import query_tools as kbt # noqa: E402
from kb_framework import vault as kbv # noqa: E402
from kb_framework import instructions as kbi # noqa: E402
def golden_graph() -> dict:
return json.loads((FIXTURES / "graph_golden.json").read_text(encoding="utf-8"))
def golden_stdio() -> dict:
return json.loads((FIXTURES / "stdio_golden.json").read_text(encoding="utf-8"))
class TestBuildGraphInMemory(unittest.TestCase):
def test_matches_golden_without_touching_disk(self) -> None:
result = kbg.build_graph(VAULT)
self.assertEqual(result.graph["nodes"], golden_graph()["nodes"])
self.assertEqual(result.graph["edges"], golden_graph()["edges"])
self.assertEqual(result.graph["version"], 1)
self.assertTrue(result.graph["generated_at"])
self.assertFalse((VAULT / "index").exists())
def test_reports_no_problems_for_clean_fixture(self) -> None:
result = kbg.build_graph(VAULT)
self.assertEqual(result.unresolved, [])
self.assertEqual(result.wiki_uses, [])
self.assertEqual(kbg.validation_problems(result), [])
def test_validation_reports_unresolved_and_bad_types(self) -> None:
import tempfile
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
(root / "a.md").write_text(
"---\nid: a\ntype: descision\n---\n[x](missing){\"type\":\"bogus\"}\n[[legacy]]\n",
encoding="utf-8",
)
result = kbg.build_graph(root)
problems = "\n".join(kbg.validation_problems(result))
self.assertIn("missing", problems)
self.assertIn("descision", problems)
self.assertIn("legacy", problems)
self.assertFalse(result.ok)
class TestQueryFunctions(unittest.TestCase):
def setUp(self) -> None:
self.graph = golden_graph()
self.stdio = golden_stdio()
def strip(self, payload: dict) -> dict:
payload = dict(payload)
payload.pop("generated_at", None)
return payload
def test_summary_matches_stdio_golden(self) -> None:
case = self.stdio["graph_summary"]
payload = kbq.summary(self.graph, **case["args"])
self.assertEqual(self.strip(payload), case["structuredContent"])
def test_search_matches_stdio_golden(self) -> None:
case = self.stdio["graph_search_nodes"]
self.assertEqual(kbq.search_nodes(self.graph, case["args"]), case["structuredContent"])
def test_neighbors_matches_stdio_golden(self) -> None:
case = self.stdio["graph_neighbors"]
self.assertEqual(kbq.neighbors(self.graph, **case["args"]), case["structuredContent"])
def test_neighbors_unknown_id_raises(self) -> None:
with self.assertRaises(ValueError):
kbq.neighbors(self.graph, "nope")
def test_node_path_resolves_id(self) -> None:
self.assertEqual(kbq.node_path(self.graph, "area-ops"), "02_areas/ops.md")
with self.assertRaises(ValueError):
kbq.node_path(self.graph, "nope")
class TestVaultHelpers(unittest.TestCase):
def test_resolve_in_root_rejects_escape_and_absolute(self) -> None:
for bad in ("../x.md", "/etc/passwd", "", " "):
with self.subTest(bad=bad), self.assertRaises(ValueError):
kbv.resolve_in_root(VAULT, bad)
self.assertEqual(kbv.resolve_in_root(VAULT, "README.md"), VAULT / "README.md")
def test_read_note_payload_text_and_binary(self) -> None:
text = kbv.read_note_payload(VAULT, "02_areas/ops.md")
self.assertEqual(text["binary"], False)
self.assertIn("# Area Ops", text["content"])
binary = kbv.read_note_payload(VAULT, "03_resources/demo/raw/paper.pdf")
self.assertTrue(binary["binary"])
self.assertEqual(set(binary), {"path", "binary", "bytes", "sha256", "hint"})
with self.assertRaises(FileNotFoundError):
kbv.read_note_payload(VAULT, "nope.md")
class TestQueryTools(unittest.TestCase):
"""The five read-only tools, as the stdio server and the connector expose them."""
def setUp(self) -> None:
self.graph = golden_graph()
self.stdio = golden_stdio()
def test_definitions_match_stdio(self) -> None:
self.assertEqual(kbt.definitions(), self.stdio["tool_definitions"])
def test_call_returns_stdio_result_shape(self) -> None:
for key in ("graph_summary", "graph_search_nodes", "graph_neighbors", "read_note_by_id", "read_note_by_path"):
case = self.stdio[key]
with self.subTest(tool=key):
result = kbt.call(case["tool"], case["args"], graph=self.graph, root=VAULT)
sc = dict(result["structuredContent"])
sc.pop("generated_at", None)
self.assertEqual(sc, case["structuredContent"])
self.assertEqual(result["content"][0]["type"], "text")
self.assertEqual(json.loads(result["content"][0]["text"]).get("path", sc.get("path")), sc.get("path"))
def test_read_note_requires_exactly_one_selector(self) -> None:
for args in ({}, {"id": "area-ops", "path": "README.md"}):
with self.subTest(args=args), self.assertRaises(ValueError):
kbt.call("read_note", args, graph=self.graph, root=VAULT)
def test_unknown_tool_raises(self) -> None:
with self.assertRaises(ValueError):
kbt.call("write_note", {}, graph=self.graph, root=VAULT)
class TestInstructions(unittest.TestCase):
def test_reads_framework_documents(self) -> None:
payload = kbi.read_instructions()
self.assertEqual([d["path"] for d in payload["documents"]], ["AGENTS.md", "AGENT_QUERY_GUIDE.md"])
self.assertIn("AGENT_QUERY_GUIDE", payload["documents"][0]["content"])
self.assertTrue(Path(payload["framework_root"]).is_dir())
if __name__ == "__main__":
unittest.main()
+3 -10
View File
@@ -40,9 +40,7 @@ class ToolTestCase(unittest.TestCase):
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name).resolve() / "vault"
self.root.mkdir()
self.state = mcp.AppState(
root=self.root, graph_path=self.root / "index" / "graph.json"
)
self.state = mcp.AppState(root=self.root)
_git(self.root, "init", "-q", "-b", "main")
_git(self.root, "config", "user.email", "test@example.com")
_git(self.root, "config", "user.name", "Test")
@@ -440,18 +438,13 @@ class TestGraphSummary(ToolTestCase):
"""
def write_graph(self, nodes: list) -> None:
self.state.graph_path.parent.mkdir(parents=True, exist_ok=True)
self.state.graph_path.write_text(
json.dumps(
{
# The graph lives in memory; seed the cache the way a build would.
self.state.graph_cache = {
"version": 1,
"generated_at": "2026-09-14T00:00:00Z",
"nodes": nodes,
"edges": [],
}
),
encoding="utf-8",
)
def node(self, node_id: str, **fields) -> dict:
base = {