Knowledge base hub — PARA-structured company memory combining company-kb and kb for persistent context, project documentation, and agent recall across sessions. Use when working with knowledge base, company knowledge, or persistent memory.
Install with Codex or Claude Copy this prompt, paste it into Codex, Claude, or another assistant, and let it review the skill page and install it for you.
A direct command skips the review prompt. Inspect the source before running it.
Knowledge base hub — PARA-structured company memory combining company-kb and kb for persistent context, project documentation, and agent recall across sessions. Use when working with knowledge base, company knowledge, or persistent memory.
{"name":"Knowledge Base Steward","title":"Master of Organizational Memory","expertise":["Knowledge Management","PARA Method","Agent Context Engineering","Information Architecture"],"philosophy":"Memory without structure is noise. Structure without recall is a graveyard.","credentials":["Knowledge systems architect","Information retrieval specialist","Multi-session context engineer"],"principles":["Write once, retrieve forever","Structure for the next agent, not for today","Decay is real — refresh or retire","Every session starts with context, never blank"]}
version
1.0.0
Core Knowledge Base Hub — Company Memory & Recall
When Not to Use
Simple or one-off tasks — if the task is straightforward, direct execution is faster than structured methodology.
Already established workflows — follow existing team conventions rather than introducing new frameworks.
When automation overhead exceeds benefit — for very small scopes, the setup cost may not be justified.
Dependencies
Python 3.8+ or Node.js 18+
Access to relevant APIs/services for your specific use case
Basic understanding of the domain concepts
Commands
# Refer to the skill's usage section for specific commands# Adapt these to your workflow
Money-Making Overview
A well-structured knowledge base is the single highest-leverage investment for an autonomous agent ecosystem. Every hour spent organizing knowledge saves 10 hours of re-discovery and context-switching. Direct revenue impact:
Capability
ROI Impact
Timeline
Session-start context loading
Eliminates 10-15 min of re-orientation per session
Day 1
Company knowledge recall
Instant access to strategies, playbooks, finance data
Total addressable value: A mature knowledge base turns a 1-person operation into a 10-person operation by eliminating context loss. Every lost context costs $50-200 in re-discovery time. With 5+ sessions per day, that's $250-1,000/day saved.
Phase 1: Initialize Knowledge Base Structure (10 min)
# 1. Verify PARA directory structure existsls -la ~/kb/projects/
ls -la ~/kb/areas/
ls -la ~/kb/resources/
ls -ka ~/kb/archives/
# 2. Read today's context — loads session memory
kb read --today
# Expected: daily note with yesterday's summary, current tasks, open decisions# 3. Quick scan of company entities
company-kb list-entities --type all
company-kb list-entities --type client
company-kb list-entities --type product
Phase 2: Capture Current Session Context (15 min)
import os
from datetime import datetime
from pathlib import Path
import yaml
KB_ROOT = Path(os.environ.get("KB_ROOT", "~/kb")).expanduser()
defensure_para_dirs():
"""Create PARA directory structure if missing."""for category in ["projects", "areas", "resources", "archives"]:
(KB_ROOT / category).mkdir(parents=True, exist_ok=True)
print("PARA structure ready")
defwrite_daily_note(date: datetime = None):
"""Write a session-start daily note with today's plan."""
date = date or datetime.now()
daily_dir = KB_ROOT / "areas" / "daily"
daily_dir.mkdir(parents=True, exist_ok=True)
note_path = daily_dir / f"{date.strftime('%Y-%m-%d')}.md"if note_path.exists():
# Load existing and append session
existing = note_path.read_text()
note_path.write_text(f"{existing}\n\n## Session {date.strftime('%H:%M')}\n- ")
print(f"Appended to existing note: {note_path}")
else:
content = f"""# Daily Note: {date.strftime('%Y-%m-%d')}
## Today's Focus
-
## Tasks
- [ ]
## Open Decisions
-
## Follow-ups
-
## Notes
-
---
*Auto-generated by KB Hub at {date.isoformat()}*
"""
note_path.write_text(content)
print(f"Created daily note: {note_path}")
# Initialize
ensure_para_dirs()
write_daily_note()
Phase 3: Store Atomic Knowledge Facts (15 min)
import yaml
defstore_fact(category: str, entity: str, fact: dict):
"""Store a structured fact as YAML in the knowledge base.
category: 'projects', 'areas', 'resources'
entity: name of the thing this fact describes
fact: dict with keys like 'type', 'value', 'source', 'date', 'status'
"""# Determine file pathif category == "projects":
dir_path = KB_ROOT / "projects" / entity
elif category == "areas":
dir_path = KB_ROOT / "areas" / entity
else:
dir_path = KB_ROOT / "resources" / entity
dir_path.mkdir(parents=True, exist_ok=True)
file_path = dir_path / "facts.yaml"# Load existing or create
facts = []
if file_path.exists():
withopen(file_path) as f:
facts = yaml.safe_load(f) or []
# Append new fact
facts.append({
**fact,
"recorded_at": datetime.now().isoformat(),
"recorded_by": "kb-hub"
})
withopen(file_path, "w") as f:
yaml.dump(facts, f, default_flow_style=False, sort_keys=False)
print(f"Stored fact in {file_path}")
return file_path
# Examples — store immediately useful facts
store_fact("resources", "mcp-clients", {
"type": "capability",
"value": "MCP client hub connects agents to 50+ servers",
"status": "active",
"priority": "high"
})
store_fact("areas", "pricing-strategy", {
"type": "decision",
"value": "Standard tier at $49/mo, Pro tier at $99/mo",
"date": "2026-07-01",
"status": "confirmed"
})
store_fact("projects", "website-redesign", {
"type": "milestone",
"value": "Design review completed, dev started",
"status": "in-progress",
"target": "2026-08-01"
})
Phase 4: Query and Recall (20 min)
defquery_kb(search_term: str, max_results: int = 10):
"""Full-text search across entire knowledge base."""import subprocess
results = []
query = subprocess.run(
["rg", "-l", "-i", search_term, str(KB_ROOT)],
capture_output=True, text=True, timeout=10
)
for file_path in query.stdout.strip().split("\n"):
ifnot file_path:
continue# Extract a snippet
snippet = subprocess.run(
["rg", "-i", "-m", "3", search_term, file_path],
capture_output=True, text=True, timeout=5
)
rel_path = Path(file_path).relative_to(KB_ROOT)
results.append({
"file": str(rel_path),
"snippet": snippet.stdout.strip()[:300],
"lines": snippet.stdout.count("\n") + 1
})
iflen(results) >= max_results:
breakifnot results:
print(f"No results for '{search_term}'")
return []
print(f"Found {len(results)} results for '{search_term}':")
for r in results:
print(f" 📄 {r['file']}")
print(f" {r['snippet']}")
print()
return results
defread_entity(name: str, category: str = None):
"""Read all knowledge about a specific entity."""
paths = []
if category:
paths = [KB_ROOT / category / name]
else:
# Search all categoriesfor cat in ["projects", "areas", "resources"]:
p = KB_ROOT / cat / name
if p.exists():
paths.append(p)
ifnot paths:
print(f"No entity found: {name}")
returnfor p in paths:
print(f"\n{'='*60}")
print(f"Entity: {p.relative_to(KB_ROOT)}")
print(f"{'='*60}")
if p.is_dir():
for f insorted(p.rglob("*")):
if f.is_file() and f.suffix in (".md", ".yaml", ".yml"):
print(f"\n--- {f.name} ---")
print(f.read_text()[:500])
iflen(f.read_text()) > 500:
print("... (truncated, use `kb read` for full)")
elif p.is_file():
print(p.read_text()[:1000])
# Query examples
query_kb("pricing")
query_kb("MCP")
read_entity("mcp-clients", "resources")
Concrete Action Flow
Every Session: KB Warmup Flow
defsession_warmup():
"""Run at session start — loads context into agent memory."""# 1. Load yesterday's daily note
yesterday = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")
note_path = KB_ROOT / "areas" / "daily" / f"{yesterday}.md"if note_path.exists():
print(f"Yesterday's context:\n{note_path.read_text()[:1000]}")
# 2. Load open decisions and pending items
decisions_file = KB_ROOT / "resources" / "decisions.yaml"if decisions_file.exists():
withopen(decisions_file) as f:
decisions = yaml.safe_load(f) or []
open_decisions = [d for d in decisions if d.get("status") == "open"]
if open_decisions:
print(f"\nOpen decisions ({len(open_decisions)}):")
for d in open_decisions:
print(f" - {d['value']}")
# 3. Check for stale content (auto-decay check)
stale = check_for_stale_content(days_old=90)
if stale:
print(f"\nStale content to review: {len(stale)} items")
# 4. Write today's session note
write_daily_note()
print("\nSession warmup complete. KB is loaded.")
Writing Flow: Capture → Structure → Store → Cross-Reference
defcapture_knowledge(entity: str, category: str, content: dict):
"""Full knowledge capture pipeline."""# Step 1: Store raw fact
fact_path = store_fact(category, entity, {
"type": content.get("type", "note"),
"value": content["value"],
"status": content.get("status", "draft"),
"priority": content.get("priority", "medium")
})
# Step 2: Write detailed markdown if neededif"detail"in content:
detail_dir = KB_ROOT / category / entity
detail_path = detail_dir / f"{entity}-details.md"withopen(detail_path, "w") as f:
f.write(f"# {entity}\n\n{content['detail']}\n")
print(f"Detail written: {detail_path}")
# Step 3: Update entity index
index_file = KB_ROOT / "resources" / "entity-index.yaml"if index_file.exists():
withopen(index_file) as f:
index = yaml.safe_load(f) or {}
else:
index = {}
if entity notin index:
index[entity] = {"category": category, "facts": [], "related": []}
index[entity]["facts"].append(content["value"])
index[entity]["last_updated"] = datetime.now().isoformat()
# Cross-reference with related entitiesif"related"in content:
index[entity]["related"].extend(content["related"])
index[entity]["related"] = list(set(index[entity]["related"]))
withopen(index_file, "w") as f:
yaml.dump(index, f, default_flow_style=False, sort_keys=False)
print(f"Entity index updated for '{entity}'")
return fact_path
Retrieval Flow: Query → Filter → Rank → Present
defretrieve_knowledge(query: str, filters: dict = None, top_k: int = 5):
"""Structured retrieval pipeline."""# Phase 1: Full-text search
results = []
import subprocess
grep_out = subprocess.run(
["rg", "-l", "-i", query, str(KB_ROOT)],
capture_output=True, text=True, timeout=10
)
files = [f for f in grep_out.stdout.strip().split("\n") if f]
# Phase 2: Apply filtersif filters:
if filters.get("category"):
files = [f for f in files iff"/{filters['category']}/"in f]
if filters.get("entity"):
files = [f for f in files if filters["entity"] in f]
if filters.get("after"):
after_ts = datetime.fromisoformat(filters["after"])
files = [f for f in files
if datetime.fromtimestamp(Path(f).stat().st_mtime) > after_ts]
# Phase 3: Rank by freshness + relevance
scored = []
for f in files:
path = Path(f)
mtime = path.stat().st_mtime
# Count query hits for relevance score
hits = subprocess.run(
["rg", "-c", "-i", query, str(path)],
capture_output=True, text=True, timeout=5
)
hit_count = int(hits.stdout.strip() or0)
freshness_score = min(1.0, (datetime.now().timestamp() - mtime) / 86400 / 30)
scored.append((hit_count * 10 + (1 - freshness_score) * 5, f))
scored.sort(key=lambda x: x[0], reverse=True)
# Phase 4: Presentfor score, f in scored[:top_k]:
rel = Path(f).relative_to(KB_ROOT)
snippet = subprocess.run(
["rg", "-i", "-m", "5", query, str(f)],
capture_output=True, text=True, timeout=5
)
print(f"[Score: {score:.1f}] {rel}")
print(f" {snippet.stdout.strip()[:200]}")
print()
return [{"file": Path(f).relative_to(KB_ROOT).as_posix(), "score": s}
for s, f in scored[:top_k]]
Decay & Archival Flow
defcheck_for_stale_content(days_old: int = 90):
"""Find knowledge that hasn't been touched in days_old days."""from datetime import timedelta
cutoff = datetime.now() - timedelta(days=days_old)
stale = []
for category in ["projects", "areas", "resources"]:
for f in (KB_ROOT / category).rglob("*.md"):
if datetime.fromtimestamp(f.stat().st_mtime) < cutoff:
stale.append(f)
return stale
defarchive_stale_content(dry_run: bool = True):
"""Move stale content to archives/ for preservation without clutter."""
stale = check_for_stale_content(90)
archive_dir = KB_ROOT / "archives"
archive_dir.mkdir(exist_ok=True)
for path in stale:
rel = path.relative_to(KB_ROOT)
archive_path = archive_dir / rel
if dry_run:
print(f"[DRY RUN] Would archive: {rel} -> {archive_path}")
else:
archive_path.parent.mkdir(parents=True, exist_ok=True)
path.rename(archive_path)
print(f"Archived: {rel}")
ifnot stale:
print("No stale content found.")
return stale if dry_run elseNone
Anti-Rationalization Table
Rationalization
Reality
"I'll remember what we decided last week"
You won't. Neither will the next agent. Write it down or lose it.
"The conversation is the memory"
Conversations disappear when the session ends. KB persists.
"Structuring by PARA is overkill for one person"
PARA scales from 1 to 100 agents. Start structured or refactor later at 10x cost.
"I'll add facts later when I have time"
"Later" never comes. Capture at the moment of decision.
"Search is good enough for retrieval"
Search finds strings. Structure finds meaning, decisions, and relationships.
"Archiving loses information"
Archiving moves stale data out of active context — it is not deletion.
"Only humans need documentation"
Agents need documentation more than humans — they have no intuition to fill gaps.
Output Format
When using the knowledge base hub, produce structured recall results:
{"query":"pricing decision","results":[{"file":"areas/pricing-strategy/facts.yaml","relevance_score":85.0,"snippet":"Decision: Standard tier at $49/mo, Pro tier at $99/mo","last_updated":"2026-07-01T14:30:00Z","status":"confirmed"}],"total_found":3,"total_returned":2,"filters_applied":{"category":"areas"}}
