
Neo4j Getting Started Skill
- 414 installs
- 101 repo stars
- Updated August 3, 2026
- neo4j-contrib/neo4j-skills
neo4j-getting-started-skill is an official Neo4j agent skill that runs an eight-stage zero-to-app pipeline for developers starting graph database projects with Aura provisioning, modeling, loading, and exploration.
About
neo4j-getting-started-skill is the official Neo4j contributor skill that orchestrates a zero-to-running-app pipeline across eight ordered stages: prerequisites, context, provision, model, load, explore, query, and build. Each stage reads its own reference file and supports both human-in-the-loop and fully autonomous operation, with documented time budgets of 15 minutes or less autonomous and up to 90 minutes with HITL. Developers reach for neo4j-getting-started-skill when starting a new Neo4j project from scratch, provisioning Aura, generating synthetic data, building a notebook or application, or running the full onboarding pipeline. The skill explicitly does not cover standalone Cypher authoring—that belongs to neo4j-cypher-skill—nor driver upgrades or CLI admin on existing databases. It produces a provisioned graph, loaded dataset, explored schema, working queries, and a runnable app or notebook artifact.
- 8 ordered stages: prerequisites → context → provision → model → load → explore → query → build
- Loads one reference file per stage only—not the entire doc set at once
- Supports HITL (≤90 min) and fully autonomous runs (≤15 min time budget)
- Integrates Neo4j MCP (read/write Cypher, schema, GDS) and data-modeling validate/visualize tools
- Explicitly excludes deep Cypher authoring, driver migration, and CLI admin—points to sibling Neo4j skills
Neo4j Getting Started Skill by the numbers
- 414 all-time installs (skills.sh)
- +29 installs in the week ending Aug 4, 2026 (Skillselion tracking)
- Ranked #141 of 911 Databases skills by installs in the Skillselion catalog
- Security screen: HIGH risk (skills.sh audit)
- Data as of Aug 4, 2026 (Skillselion catalog sync)
npx skills add https://github.com/neo4j-contrib/neo4j-skills --skill neo4j-getting-started-skillAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 414 |
|---|---|
| repo stars | ★ 101 |
| Security audit | 2 / 3 scanners passed |
| Last updated | August 3, 2026 |
| Repository | neo4j-contrib/neo4j-skills ↗ |
How do you bootstrap a new Neo4j project?
Run the official eight-stage Neo4j pipeline from prerequisites through provision, model, load, explore, query, and build a runnable app or notebook.
Who is it for?
Backend developers starting a greenfield Neo4j graph project who need guided provisioning, modeling, data load, and first application scaffolding.
Skip if: Teams maintaining existing Neo4j deployments who only need Cypher tuning, driver migration, or CLI administration on live databases.
When should I use this skill?
The user starts a new Neo4j project, provisions Aura, needs synthetic graph data, or asks to run the full eight-stage onboarding pipeline.
What you get
Provisioned Neo4j Aura or local instance, graph data model, loaded dataset, explored graph, working Cypher queries, and a runnable app or notebook.
- provisioned graph database
- loaded dataset
- runnable app or notebook
By the numbers
- 8-stage zero-to-app pipeline
- ≤15 minute autonomous time budget
- ≤90 minute human-in-the-loop time budget
Files
Neo4j Getting-Started Skill
Guide a user or agent from zero to a working Neo4j application by executing the 8 stages below in order.
At the start of each stage: read the corresponding ${CLAUDE_SKILL_DIR}/references/<stage-name>.md file and follow its instructions. Only load the stage you are currently executing — not all at once.
"User" means both a human developer and an autonomous coding agent.
---
When to Use
- New Neo4j project from scratch (local/Docker/Aura)
- Full onboarding: zero → DB → model → load → app
- Generating synthetic data for demos or dev
When NOT to Use
- Cypher authoring on existing project →
neo4j-cypher-skill - Driver upgrades / Cypher migration →
neo4j-migration-skill - Admin on existing DB (backup, restore, import) →
neo4j-cli-tools-skill
---
Project Structure
All generated code, data, scripts, queries, and notebooks must be written to the working directory so the user can inspect, reuse, and re-run them after the session ends. Never generate output only as text in the conversation — always write it to a file.
Organize files into this layout. Create subdirectories before writing files.
.env ← DB credentials (gitignored, loaded by python-dotenv)
aura.env ← Aura API credentials (gitignored, never overwrite)
progress.md ← stage-by-stage progress (this skill writes it)
requirements.txt ← Python dependencies
schema/
schema.json ← graph model definition
schema.cypher ← DDL: constraints + indexes
reset.cypher ← wipe all data (keep schema)
data/
generate.py ← synthetic data generator (DATA_SOURCE=synthetic)
import.py ← CSV/file importer (DATA_SOURCE=csv or relational)
*.csv ← any provided or generated data files
queries/
queries.cypher ← validated Cypher query library
scripts/
provision_aura.py ← Aura provisioning script (generated during provision stage)
notebook.ipynb ← app artifact (root — standard jupyter convention)
app.py ← app artifact (root — streamlit run app.py)
main.py ← app artifact (root — uvicorn main:app)
graphrag_app.py ← app artifact (root)Root-level files (.env, requirements.txt, app code) stay at root because tooling expects them there. Everything else goes in the appropriate subfolder.
---
Progress Tracking
The skill maintains progress.md in the working directory to support resumability.
On startup: 1. Check if progress.md exists. 2. If it exists, find the first pending stage:
grep -B1 "^status: pending" progress.md | grep "^###" | head -13. Resume from that stage. Read its context block (the key=value lines beneath the header) to restore DOMAIN, USE_CASE, NEO4J_URI, etc. — do not re-ask the user for information already recorded. 4. For each completed stage, read every file listed in its files= line before proceeding. These files are the ground truth — do not reconstruct their content from memory.
schema/schema.json→ re-read before model, load, query, or build stagesqueries/queries.cypher→ re-read before build stagedata/generate.py→ re-read before import or reset
5. If progress.md does not exist, start from 0-prerequisites.
On stage completion — update (or create) progress.md:
- If the stage's
###section already exists, updatestatus: pending→status: doneand append any new key=value lines. - If the section doesn't exist, append it following the format below.
Format:
# Neo4j Getting-Started — Progress
<!-- Resume: grep for "status: pending" to find the next stage -->
### 0-prerequisites
status: done
### 1-context
status: done
DOMAIN=social
USE_CASE=friend recommendations
EXPERIENCE=beginner
DB_TARGET=aura-free
DATA_SOURCE=synthetic
APP_TYPE=notebook
EXEC_METHOD=query-api
### 2-provision
status: done
NEO4J_URI=neo4j+s://abc123.databases.neo4j.io
### 3-model
status: done
labels=Person,Post
relationships=FOLLOWS,POSTED
constraints=2
### 4-load
status: done
nodes=200 Person, 50 Post
relationships=1400 FOLLOWS, 300 POSTED
### 5-explore
status: pending
### 6-query
status: pending
### 7-build
status: pending---
Execution Protocol
For each stage: 1. Announce the stage: "## Stage: <name> — <purpose>" 2. Read ${CLAUDE_SKILL_DIR}/references/<name>.md 3. Execute the instructions in that file 4. Verify the stage's completion condition 5. Update progress.md with status: done and stage-specific context 6. Proceed to the next stage (HITL: pause for approval first)
If a stage fails, recover using the error guidance in the stage reference file. Do not skip stages unless the skip condition below explicitly permits it.
---
Stages
Stages run in the numbered order shown. Each depends on the one before it completing successfully (except where a skip condition applies). Read the linked reference file when entering each stage.
0-prerequisites → 1-context → 2-provision → 3-model → 4-load → 5-explore → 6-query → 7-buildShared capabilities used across multiple stages:
- Cypher execution:
${CLAUDE_SKILL_DIR}/references/capabilities/execute-cypher.md(3 options;EXEC_METHODchosen incontext) - Cypher authoring rules:
${CLAUDE_SKILL_DIR}/references/capabilities/cypher-authoring.md(or defer toneo4j-cypher-authoring-skill) - MCP configuration:
${CLAUDE_SKILL_DIR}/references/capabilities/mcp-config.md(used inprerequisitesandbuild) - Query validation:
${CLAUDE_SKILL_DIR}/scripts/validate_queries.py— batch-validate all queries in one call (used inquery)
---
0 — prerequisites
Purpose: Verify and install required CLI tools before doing anything else. Reference: ${CLAUDE_SKILL_DIR}/references/0-prerequisites.md Completes when: neo4j-mcp binary is reachable; .gitignore has .env entry. Never skip.
---
1 — context
Purpose: Collect domain, use-case, experience, infrastructure target, data source, and output type. Detect EXEC_METHOD for Cypher execution. Reference: ${CLAUDE_SKILL_DIR}/references/1-context.md Completes when: DOMAIN, USE_CASE, EXPERIENCE, DB_TARGET, DATA_SOURCE, APP_TYPE, EXEC_METHOD are known. Skip condition: all variables already provided in conversation context.
---
2 — provision
Purpose: Provision a running Neo4j database and save credentials to .env. Reference: ${CLAUDE_SKILL_DIR}/references/2-provision.md Completes when: .env exists with NEO4J_URI/USERNAME/PASSWORD/DATABASE; connectivity verified. Skip condition: DB_TARGET=existing → write .env from user credentials, proceed to 3-model.
---
3 — model
Purpose: Design or discover a graph data model suited to the use-case. Reference: ${CLAUDE_SKILL_DIR}/references/3-model.md Completes when: schema.json and schema.cypher written. Skip condition: DATA_SOURCE=demo → use demo schema, proceed to 4-load. HITL checkpoint (HITL mode only — skip entirely in autonomous mode): show model draft, wait for approval.
---
4 — load
Purpose: Apply schema constraints, then import data (demo, synthetic, CSV, or documents). Reference: ${CLAUDE_SKILL_DIR}/references/4-load.md Depends on: 3-model (constraints must exist before import). Completes when: node count ≥ 50; import/ scripts written; reset.cypher written.
---
5 — explore
Purpose: Deliver a visual entry point to the graph — the "it clicks" moment. Reference: ${CLAUDE_SKILL_DIR}/references/5-explore.md Completes when: browser URL printed to user, or notebook visualization cell added. Hard gate — never skip.
---
6 — query
Purpose: Generate and validate a Cypher query library for the use-case. Reference: ${CLAUDE_SKILL_DIR}/references/6-query.md Completes when: queries.cypher has ≥5 queries; ≥2 traversals; ≥3 return results.
---
7 — build
Purpose: Generate a runnable application, dashboard, notebook, or agent integration. Reference: ${CLAUDE_SKILL_DIR}/references/7-build.md Completes when: artifact exists, passes syntax check, returns non-empty use-case results.
---
Success Gates (all 7 required)
| Gate | Stage | Condition |
|---|---|---|
db_running | provision | driver.verify_connectivity() succeeds |
model_valid | model | ≥2 node labels, ≥1 rel type, ≥1 constraint in DB |
data_present | load | MATCH (n) RETURN count(n) ≥ 50 |
queries_work | query | ≥5 queries; ≥2 traversals; ≥3 return ≥1 result |
graph_visible | explore | Browser URL or notebook viz delivered to user |
app_generated | build | Artifact exists, passes syntax, returns non-empty results |
integration_ready | build | MCP config or agent framework code present (if requested) |
---
Fast Paths
| Situation | Action |
|---|---|
DB_TARGET=existing | Skip provision; write .env from user creds; go to model |
DATA_SOURCE=demo | Skip custom modeling; use demo schema; jump to load |
DB_TARGET=existing + data present | Skip provision, model, load; introspect schema; go to explore |
---
HITL vs Autonomous Mode
HITL (conversational): pause after model for model review; pause after load for data review.
Autonomous (CI-like, all context provided upfront): never pause for approval at any stage; auto-approve all decisions; proceed immediately through all 8 stages; print browser URL to stdout; target ≤15 min from DB running.
How to detect autonomous mode — check at the start of stage 1:
Autonomous if ANY of the following are true:
- The initial prompt contains all of:
DOMAIN,USE_CASE,EXPERIENCE,DB_TARGET,DATA_SOURCE,APP_TYPE(or equivalent phrasing like "Domain: X, use-case: Y, ...") - The session was started with
--auto-approveor similar non-interactive flag - All context variables are already recorded in
progress.md(resuming an autonomous run)
HITL if: the user opened a fresh conversation without providing full context upfront.
In autonomous mode: every HITL checkpoint in every stage reference file is automatically skipped. Do not ask for approval. Do not say "does this look right?" Do not pause. Continue to the next step immediately.
---
Final Summary (deliver after all gates pass)
Step 1 — write `README.md` to the working directory using the template below. Fill in every <placeholder> from progress.md and the actual generated files. This is a required output — do not skip it.
IMPORTANT — portable commands: All re-run commands in README.md MUST use python3 (never an absolute path like /opt/homebrew/bin/python3.14 or /usr/local/bin/python3). The README is shared with others who have different Python installs.
# <DOMAIN> Graph — <USE_CASE>
A synthetic <DOMAIN> graph built with Neo4j, covering <USE_CASE>.
Generated by the neo4j-getting-started-skill on <date>.
## What's in the graph
| Label | Count | Description |
|-------|-------|-------------|
| <Label> | <N> | <one line> |
**Relationships:** <TYPE1>, <TYPE2>, ...
**Constraints:** <N> uniqueness constraints applied
## Explore visually
Open in Neo4j Browser:
<browser_url>
Use `NEO4J_PASSWORD` from `.env` to connect, then run:// Starter query — shows the full graph sample MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 50
## Files
| File | Purpose | Re-run |
|------|---------|--------|
| `schema/schema.json` | Graph model | — |
| `schema/schema.cypher` | Constraints + indexes | `source .env && cypher-shell -a $NEO4J_URI -u $NEO4J_USERNAME -p $NEO4J_PASSWORD --file schema/schema.cypher` |
| `schema/reset.cypher` | Wipe data, keep schema | `source .env && cypher-shell -a $NEO4J_URI -u $NEO4J_USERNAME -p $NEO4J_PASSWORD --file schema/reset.cypher` |
| `data/generate.py` | Regenerate synthetic data | `source .venv/bin/activate && python3 data/generate.py` |
| `data/import.py` | Re-import CSVs into Neo4j | `source .venv/bin/activate && python3 data/import.py` |
| `queries/queries.cypher` | Query library | Paste into Neo4j Browser |
| `<artifact>` | <app type> | `<run command>` |
| `requirements.txt` | Python dependencies | `source .venv/bin/activate && pip install -r requirements.txt` |
(Omit `data/generate.py` row when `DATA_SOURCE=csv`; omit `data/import.py` row when `DATA_SOURCE=synthetic`.)
## Run the app
python3 -m venv .venv # skip if .venv already exists source .venv/bin/activate pip install -r requirements.txt <run command>
<For FastAPI only — include this section:>
Open http://localhost:8000/docs for the interactive API docs.
<For MCP integration — include this section when APP_TYPE includes mcp:>
## MCP integration
To query your graph directly from Claude:
**Claude Code** — copy `mcp-claude-code.json` into `.claude/settings.json`:cp mcp-claude-code.json .claude/settings.json
Then reload Claude Code (`/reload` or restart). Ask: "What node labels are in my Neo4j database?"
**Claude Desktop** — merge `mcp-claude-desktop.json` into
`~/Library/Application Support/Claude/claude_desktop_config.json`, then restart Claude Desktop.
Available MCP tools: `get-schema`, `read-cypher`, `write-cypher`.
## Reset and reload
source .env cypher-shell -a $NEO4J_URI -u $NEO4J_USERNAME -p $NEO4J_PASSWORD --file schema/reset.cypher source .venv/bin/activate python3 data/generate.py # or skip if using your own CSVs python3 data/import.py
## Sample queries
// <use-case-specific query 1 — fill in from queries/queries.cypher> <query>
// <use-case-specific query 2> <query>
(Cypher comments use `//`, not `--`.)
## Next steps
- Explore [GraphAcademy](https://graphacademy.neo4j.com) to deepen your Neo4j knowledge
- Edit `data/*.csv` to change the dataset, then re-run `data/import.py`
- Extend the model: add new node labels or relationship types in `schema/schema.json`Step 2 — print this to the conversation:
✓ Neo4j Getting-Started — Complete
Database: <NEO4J_URI>
Browser: https://browser.neo4j.io/?connectURL=<encoded>
── What was generated (keep these files) ───────────────────────
schema/schema.json Graph model definition
schema/schema.cypher Re-apply constraints/indexes: cypher-shell ... --file schema/schema.cypher
schema/reset.cypher Wipe data, keep schema: cypher-shell ... --file schema/reset.cypher
data/generate.py Regenerate synthetic data: source .venv/bin/activate && python3 data/generate.py
data/*.csv Source data files — edit to change the dataset
data/import.py Re-import from CSVs: source .venv/bin/activate && python3 data/import.py
queries/queries.cypher Query library — paste into Neo4j Browser or run with cypher-shell
<app-file> <run-command>
requirements.txt Install deps: source .venv/bin/activate && pip install -r requirements.txt
── Gates ───────────────────────────────────────────────────────
db_running ✓ model_valid ✓ data_present ✓ queries_work ✓
graph_visible ✓ app_generated ✓ integration_ready ✓/–
── Next steps ──────────────────────────────────────────────────
- Explore: open the Browser URL → run MATCH (n)-[r]->(m) RETURN n,r,m LIMIT 50
- Iterate: edit data/*.csv → source .venv/bin/activate && python3 data/import.py (reset first)
- Learn: https://graphacademy.neo4j.comOmit lines that don't apply (e.g. omit data/import.py when DATA_SOURCE=synthetic, omit data/generate.py when DATA_SOURCE=csv).
---
Checklist
- [ ] Prerequisites met (Docker/Python/Java; Aura API key if cloud)
- [ ] DB reachable —
RETURN 1in cypher-shell - [ ] Constraints + indexes ONLINE before data load
- [ ] Data loaded —
MATCH (n) RETURN count(n)> 0 - [ ] queries.cypher: all queries return expected results
- [ ] App/notebook runs end-to-end
- [ ]
.envgitignored; credentials not hardcoded
AGENTS.md — neo4j-getting-started-skill
Project Purpose
This skill guides a user (or coding agent) from zero to a running Neo4j application. It covers 5 stages: DB provisioning → data modeling → data import → query generation → app/integration.
Target: ≤15 min autonomous, ≤90 min HITL. Designed for Claude Code, Cursor, Windsurf.
Commands
# Run skill manually (interactive)
claude --append-system-prompt SKILL.md
# Run skill with persona (autonomous test)
uv run python3 tests/harness/runner.py --persona tests/personas/alex_beginner.yml --verbose
# Run all personas
uv run python3 tests/harness/runner.py --all-personas
# Quick connectivity test (assumes .env exists)
python3 -c "from neo4j import GraphDatabase; import os; from dotenv import load_dotenv; load_dotenv(); d=GraphDatabase.driver(os.getenv('NEO4J_URI'),auth=(os.getenv('NEO4J_USERNAME'),os.getenv('NEO4J_PASSWORD'))); d.verify_connectivity(); print('OK')"Directory Layout
neo4j-getting-started-skill/
├── SKILL.md ← Main skill (system prompt extension)
├── AGENTS.md ← This file
├── PLAN.md ← Implementation plan with step IDs
├── neo4j-getting-started-research.md ← Research doc, living reference
├── references/
│ ├── stage1-provisioning.md ← aura-cli, docker, Desktop instructions
│ ├── stage2-data-modeling.md ← model gen, DDL, arrows.app
│ ├── stage3-data-import.md ← LOAD CSV, synthetic gen, bulk import
│ ├── stage4-queries.md ← query templates by domain
│ ├── stage5-apps.md ← FastAPI, Streamlit, Express, MCP templates
│ ├── domain-patterns.md ← Pre-built models: social, ecommerce, finance, etc.
│ └── integration-patterns.md ← LangChain, LlamaIndex, CrewAI, Mastra
└── tests/
├── personas/
│ ├── alex_beginner.yml ← Persona 1: social network, Aura Free, notebook
│ ├── sam_developer.yml ← Persona 2: e-commerce, CSV, FastAPI + MCP
│ ├── jordan_ai_engineer.yml ← Persona 3: RAG/KG, GraphRAG pipeline
│ ├── morgan_analyst.yml ← Persona 4: fraud detection, Streamlit
│ └── riley_platform_engineer.yml ← Persona 5: SaaS, multi-instance
├── harness/
│ ├── runner.py ← Test executor (invokes Claude, validates gates)
│ └── validator.py ← 6-gate validation pipeline
└── results/ ← JSON + Markdown reports per runConventions
SKILL.mdfrontmatter:name,description,version,allowed-tools,compatibility- References in
references/namedstage{N}-<topic>.mdor<domain>-patterns.md - Persona YAML:
persona,inputs,expected_outputs,success_gates,test_config - All Cypher in reference files must start with
CYPHER 25 - All generated
.envfiles go in the working directory, never committed - Gate IDs:
db_running,model_valid,data_present,queries_work,app_generated,mcp_configured,time_budget
Gotchas
- Aura Free is 512MB — synthetic data should stay under 100K nodes to avoid OOM
- neo4j driver 6.x: package is
neo4j(notneo4j-driver), requires Python ≥3.10 - CYPHER 25 pragma: every generated query must start with
CYPHER 25 - aura-cli polling: use
aura-cli instance get <id> --output jsonto check status; parse.statusfield - Docker startup: wait ≥20s after
docker runbefore attempting connection - MCP config location: Claude Desktop =
~/Library/Application Support/Claude/claude_desktop_config.json(macOS); Claude Code =~/.claude/settings.jsonor.claude/settings.jsonin project - Notebook validation:
python -m json.tool notebook.ipynbchecks JSON validity;jupyter nbconvert --to scriptchecks cell syntax - Official Neo4j MCP server: binary
neo4j-mcpfrom https://github.com/neo4j/mcp. MCP tool names:get-schema,read-cypher,write-cypher,list-gds-procedures. Config:"command": "neo4j-mcp"+ envNEO4J_URI/USERNAME/PASSWORD/DATABASE. May be at$HOME/bin/neo4j-mcpor./neo4j-mcp. - Aura provisioning: prefer
aura-cli. Fallback: Aura REST API athttps://api.neo4j.io/v1/with Bearer token fromhttps://api.neo4j.io/oauth2/token. Do NOT usemcp-neo4j-cloud-aura-api. - Connectivity check options: (1) Python
driver.verify_connectivity(), (2)cypher-shell -a ... "RETURN 1", (3) Neo4j Query API (HTTP):POST /db/<db>/query/v2with{"statement": "RETURN 1"}and Basic Auth. Use HTTP fallback when neither driver nor cypher-shell available. - neo4j-rust-ext: always add to Python
requirements.txt—neo4j-rust-ext>=0.0.1alongsideneo4j>=6.0.0. - Password hard stop: write
.envimmediately from provisioning JSON response. Hard stop + tell user to verifygrep PASSWORD .env. - MERGE order: MERGE all nodes first (per label), then MERGE relationships. Never MERGE a rel before its endpoint nodes exist.
- Time budget: 15-min clock starts after DB is RUNNING. Use provisioning wait for Stage 2 design.
- GDS availability: Aura Free has NO GDS. Check with
CALL gds.version()orlist-gds-proceduresMCP tool before generating GDS queries. - Query gate: ≥2 traversal queries required (relationship pattern in MATCH). Count-only queries do not satisfy.
- Graph visibility gate (hard): must print standalone browser URL or add notebook viz cell.
- App gate: app must return non-empty results for the use-case question, not just compile.
- reset.cypher: always generate — allows clean re-runs of import scripts.
- Schema first: always call
get-neo4j-schemaorCALL db.schema.visualization()before generating Cypher - MERGE not CREATE: all data generation must use MERGE for idempotency
- Parameter defaults for testing: when executing queries with
$param, substitute defaults (LIMIT 20, string'test') in the validator
Related Repos
/Users/mh/d/llm/neo4j-skills/— parent skill collection, test harness patterns/Users/mh/d/llm/aura-onboarding-assistant/— UI onboarding app (10-phase spec inspec/spec.md)/Users/mh/d/llm/mcp-neo4j/— MCP servers: cypher, cloud-aura-api, data-modeling, memory/Users/mh/d/llm/neo4j-mcp/— standalone Aura MCP server
Success Gate Definitions
| Gate | What it checks | Pass condition |
|---|---|---|
db_running | DB connectivity | driver.verify_connectivity() no exception |
model_valid | Schema completeness | ≥2 node labels, ≥1 relationship type |
data_present | Data loaded | MATCH (n) RETURN count(n) ≥ persona min_nodes |
queries_work | Query correctness | ≥3 of 5 queries return ≥1 row |
app_generated | App file exists + valid | File exists + syntax check passes |
mcp_configured | MCP connected | Settings JSON has neo4j server entry |
time_budget | Within time limit | elapsed ≤ timeout_seconds from persona YAML |
Dependencies (pyproject.toml — to be created)
[project]
name = "neo4j-getting-started-skill-tests"
requires-python = ">=3.10"
dependencies = [
"neo4j>=6.0.0",
"pyyaml>=6.0",
"python-dotenv>=1.0.0",
]neo4j-getting-started-skill
A Claude Code skill for the complete Neo4j getting-started journey — from zero to a running graph application in one session.
Prerequisites
- Claude Code (CLI or IDE extension)
- Python ≥ 3.10 (
python3 --version) - Docker — only required if
db_target=local-docker - Aura API credentials in
aura.env— only required ifdb_target=aura-freeoraura-pro
Installation
# From the repo root
make install-skillOr manually:
cp -R . ~/.claude/skills/neo4j-getting-started-skill/Usage
Start any Claude Code session in an empty directory and trigger the skill:
/neo4j-getting-started-skillAny context you include is extracted from natural language — no special syntax needed. Stage 1 asks for anything that's still missing, so all context is optional upfront:
/neo4j-getting-started-skill I want to build a friend recommendation system. I'm a beginner./neo4j-getting-started-skill Healthcare patient journey analysis, intermediate Python dev,
running on local Docker, generate synthetic data, Jupyter notebook please./neo4j-getting-started-skill fraud detection for a fintech startupAutonomous mode kicks in when all 6 variables (domain, use-case, experience, db target, data source, app type) can be inferred from your message — the skill then runs all 8 stages without stopping. If any are missing, stage 1 asks for them conversationally before proceeding.
What happens
The skill runs 8 stages in order:
| Stage | What it does |
|---|---|
0-prerequisites | Downloads neo4j-mcp binary, creates .venv, sets up .gitignore |
1-context | Collects domain, use-case, experience, DB target, data source, app type |
2-provision | Provisions or connects to a Neo4j database; writes .env |
3-model | Designs a graph schema; writes schema.json + schema.cypher |
4-load | Applies constraints; loads demo/synthetic/CSV/document data |
5-explore | Opens Neo4j Browser for visual exploration; runs neo4j-viz preview |
6-query | Generates and validates a Cypher query library (queries/queries.cypher) |
7-build | Generates a runnable app; installs dependencies into .venv |
Database targets (db_target)
| Value | What happens |
|---|---|
aura-free | Creates a new Aura Free instance via REST API (requires aura.env) |
aura-pro | Creates an Aura Professional instance (requires aura.env) |
local-docker | Runs neo4j:enterprise in Docker on localhost:7687 — Docker must be installed |
local-desktop | Connects to a running Neo4j Desktop instance on localhost:7687 |
existing-cloud | Connects to any existing Neo4j instance — you provide URI + password |
For local-docker, the skill pulls neo4j:enterprise, mounts a ./neo4j-data/ volume for persistence, and waits up to 90 s for Bolt to be ready. No extra setup needed beyond Docker.
Data sources (data_source)
| Value | What happens |
|---|---|
synthetic | Generates realistic fake data in Python (data/generate.py + data/import.py) |
demo | Loads a Neo4j public demo dataset (Movies, Northwind, etc.) |
csv | Imports your own CSV files from data/ |
documents | Builds a knowledge graph from documents using neo4j-graphrag (GraphRAG path) |
App types (app_type)
| Value | Output | Run command |
|---|---|---|
notebook | notebook.ipynb with schema, viz, and use-case cells | .venv/bin/jupyter notebook notebook.ipynb |
streamlit | app.py dashboard with sidebar controls + graph viz | .venv/bin/streamlit run app.py |
fastapi | main.py REST API with /health + use-case endpoints | .venv/bin/uvicorn main:app --reload |
graphrag | graphrag_app.py hybrid vector+graph retrieval pipeline | .venv/bin/python3 graphrag_app.py |
mcp | .claude/settings.json MCP server config for Claude Code | restart Claude Code |
explore-only | queries/queries.cypher + README.md only — no app | — |
Modes
HITL (human-in-the-loop, default): pauses at key checkpoints — after schema design and after data load — for user review before proceeding.
Autonomous: when all 6 context variables (domain, use_case, experience, db_target, data_source, app_type) are present in the initial prompt, the skill skips all HITL pauses and runs end-to-end without interruption. Target completion: ≤15 min (local Docker) or ≤25 min (Aura provisioning included).
Resumability
The skill writes progress.md after each stage. If a session is interrupted, invoke the skill again in the same directory — it reads progress.md, finds the first status: pending stage, and resumes from there without re-asking questions.
Files produced
.env ← DB connection credentials (gitignored)
progress.md ← stage-by-stage progress log
requirements.txt ← Python dependencies
README.md ← generated project README
.venv/ ← Python virtual environment (gitignored)
schema/
schema.json ← graph model definition
schema.cypher ← DDL: constraints + indexes
reset.cypher ← wipe data, keep schema
data/
generate.py ← synthetic data generator (data_source=synthetic)
import.py ← CSV/file importer (data_source=csv)
*.csv ← data files
queries/
queries.cypher ← validated Cypher query library (≥5 queries, ≥3 traversal)
notebook.ipynb ← (app_type=notebook)
app.py ← (app_type=streamlit)
main.py ← (app_type=fastapi)
graphrag_app.py ← (app_type=graphrag)
.claude/settings.json ← (app_type=mcp or integration=mcp)Input credential files (not generated by the skill — you provide these):
aura.env ← Aura API credentials (gitignored); see references/2-provision.mdMCP integration
When app_type=mcp or integration=mcp, the skill writes a neo4j MCP server config pointing at the local neo4j-mcp binary. After restarting Claude Code you can ask questions about your graph in natural language: "What node labels exist?", "Show me the top 10 patients by encounter count", etc.
Available MCP tools: get-schema, read-cypher, write-cypher, list-gds-procedures.
Running tests
# Single persona
make integration-elena # local Docker — healthcare notebook
make integration-alex # Aura — social network notebook
make integration-priya # Aura — fraud detection FastAPI
# All personas (sequential)
python3 neo4j-getting-started-skill-tests/harness/runner.py --all-personas
# Keep Docker container after run (for manual inspection)
make integration-elena # container kept running by default
# Stop it later:
docker stop neo4j-elena-test && docker rm neo4j-elena-testTest results land in neo4j-getting-started-skill-tests/results/. Each run produces a JSON gate report and a DB snapshot.
Configuration
| File | Purpose |
|---|---|
aura.env | Aura API CLIENT_ID + CLIENT_SECRET for provisioning new instances |
.env | DB connection URI, username, password, database — written by the provision stage |
See references/2-provision.md for Aura credential setup.
Stage 0 — prerequisites
Verify and install required CLI tools before anything else.
neo4j-mcp (official Neo4j MCP server — required)
MCP tool names exposed by this server:
get-schema— introspect node labels, relationship types, property keysread-cypher— execute read-only Cypherwrite-cypher— execute write Cypher (disabled in read-only mode)list-gds-procedures— list available GDS procedures (only if GDS is installed)
# Check if already installed
which neo4j-mcp 2>/dev/null || ls $HOME/bin/neo4j-mcp 2>/dev/null && echo "FOUND" || echo "MISSING"If missing, download binary:
PLATFORM=$(uname -s | tr '[:upper:]' '[:lower:]')
ARCH=$(uname -m)
[ "$ARCH" = "x86_64" ] && ARCH="amd64"
[ "$ARCH" = "aarch64" ] && ARCH="arm64"
curl -fsSL "https://github.com/neo4j/mcp/releases/latest/download/neo4j-mcp-${PLATFORM}-${ARCH}" \
-o ./neo4j-mcp && chmod +x ./neo4j-mcp
./neo4j-mcp --versionThe binary can live in ./neo4j-mcp (project-local) or $HOME/bin/neo4j-mcp (user-wide). Either works.
aura-cli (optional — for Aura provisioning via CLI)
which aura-cli 2>/dev/null && echo "FOUND" || echo "MISSING — using Aura REST API directly (no install needed)"aura-cli is not required — the provision stage uses the Aura REST API via curl as the primary path, which works without any binary.
Docker (optional — for local Docker path only)
docker --version 2>/dev/null && echo "FOUND" || echo "MISSING"Python (required for app generation)
python3 --version && which python3STOP. Use ONLY `python3` — never probe `python3.10`, `python3.11`, `python3.12`, `python3.13` etc. individually. One command, one result. Any version ≥3.10 is fine.
Python virtual environment (required — always create)
Modern Python (3.12+, all macOS system Pythons) forbids global `pip install` with the "externally-managed-environment" error. A .venv in the working directory is required for all package installs.
python3 -m venv .venv
echo "✓ Virtual environment created at .venv"All subsequent pip install, python3 script.py, jupyter, streamlit, and uvicorn commands must use this venv:
- Install:
.venv/bin/pip install ... - Run scripts:
.venv/bin/python3 script.py - Run app:
.venv/bin/jupyter notebook .../.venv/bin/streamlit run .../.venv/bin/uvicorn ...
Never use bare `pip install` or `python3` commands after this point — they may target the wrong (system) Python.
.gitignore setup (always run)
for entry in .env aura.env neo4j-mcp "mcp-*.json" .provision.lock neo4j-data/ .venv/; do
grep -qxF "$entry" .gitignore 2>/dev/null || echo "$entry" >> .gitignore
done
echo "✓ .gitignore updated"Completion condition
neo4j-mcpbinary reachable (local or on PATH).gitignorecontains.env,aura.env, and.venv/- Python ≥3.10 available
.venv/created in working directory
On Completion — write to progress.md
### 0-prerequisites
status: done
PYTHON=<path from `which python3`>
NEO4J_MCP=<path from `which neo4j-mcp` or local path>
VENV=.venvStage 1 — context
Collect the user's domain, use-case, goals, and preferences.
Autonomous mode detection — do this FIRST
Before asking anything, attempt to extract all 6 context variables from the initial prompt. Look for: domain, use-case, experience level, database target, data source, and app type — either as explicit key=value pairs or inferred from natural language.
The autonomous/HITL decision is based solely on whether all 6 variables are present — never on phrasing like "guide me", "help me", or "walk me through". Those phrases describe intent, not preference for interactive vs automated execution.
If all 6 variables can be extracted → AUTONOMOUS MODE.
- Extract and record them immediately (no questions needed)
- Record
MODE=autonomousin progress.md alongside the other variables - Never pause for approval at any HITL checkpoint in any stage
- Proceed through all 8 stages without stopping
If any variable is missing → HITL MODE.
- Ask all missing questions in one combined message (see below)
- Record
MODE=hitlin progress.md
What to ask (HITL mode only)
Combine into one conversational message — not a form dump. If any value is already known from context, skip that question.
Hi! To get you up and running with Neo4j, I need a few things:
1. What's your domain or industry?
(e.g. social network, e-commerce, finance, healthcare, logistics, media, legal, IoT, or describe your own)
2. What's the specific use-case?
Be as concrete as possible — e.g. "product recommendations", "fraud ring detection",
"GraphRAG over internal documents", "supply chain visibility"
3. Your experience with Neo4j?
beginner / intermediate / advanced
4. Where to run the database?
A) Aura Free — easiest, 14-day trial, no credit card needed ← default for beginners
B) Aura Pro / Enterprise — I have or will create an account
C) Local Docker — I have Docker installed
D) Neo4j Desktop — GUI, already installed
E) I already have a running database
If they choose Aura Pro (B), ask as a follow-up:
"Any cloud provider preference? (GCP / AWS / Azure — or leave blank and I'll pick the closest region to you)"
Skip this follow-up for Aura Free (GCP only) and local options.
5. What data do you have?
A) Start with a pre-built demo dataset — fastest path to first insight
B) Generate synthetic data to match my use-case
C) I have CSV files to import
D) I have a relational database (PostgreSQL / MySQL / other)
E) I have documents to build a knowledge graph / GraphRAG pipeline
(txt, md, pdf — will be ingested via SimpleKGPipeline into data/)
DATA_SOURCE inference rules:
- If the user mentions having files, documents, PDFs, contracts, papers → documents
- If files already exist in data/ → documents (even if user didn't say so)
- Only use synthetic if the user has NO data and explicitly wants generated examples
6. What do you want to build?
A) Python notebook (Jupyter / VS Code)
B) Streamlit dashboard
C) FastAPI backend
D) GraphRAG pipeline (LLM + graph retrieval)
E) Just queries + visual exploration
F) MCP server integration for an agent (Claude Desktop / Claude Code)Defaults (apply when user says "just get started" or is a beginner)
DB_TARGET = aura-free
DATA_SOURCE = demo (offer Movies as the default demo)
APP_TYPE = notebook
LANGUAGE = python
EXPERIENCE = beginnerVariables to store
DOMAIN = <domain/industry>
USE_CASE = <specific use-case description>
EXPERIENCE = beginner | intermediate | advanced
DB_TARGET = aura-free | aura-pro | local-docker | local-desktop | existing
DATA_SOURCE = demo | synthetic | csv | relational | documents
APP_TYPE = notebook | streamlit | fastapi | graphrag | explore-only | mcp
LANGUAGE = python (v1 only; javascript in phase 2)
CLOUD_PROVIDER = gcp | aws | azure | (omit if no preference)
REGION_HINT = <geographic hint inferred from user context — see below>CLOUD_PROVIDER comes from user preference (asked for Aura Pro only). REGION_HINT is inferred from geographic signals (never asked).
The provision stage combines both: user's preferred provider, closest available region within that provider using the geographic hint.
REGION_HINT — infer, don't ask
Infer from any available signal — do not ask:
- Explicit mention: "I'm in Brazil", "our servers are in Tokyo", "EU data residency required"
- Language/locale of the conversation
- Timezone in system prompt or
dateoutput (e.g.CET→ western Europe,BRT→ Brazil) - Initial prompt already contains a region → use it directly
Store the geographic area only — the provision stage maps it to the user's chosen provider:
| Signal | REGION_HINT |
|---|---|
| Western Europe / CET / BST / CEST | europe-west |
| Eastern Europe | europe-east |
| US East Coast / EST / EDT | us-east |
| US West Coast / PST / PDT | us-west |
| Brazil / BRT | sa-east |
| Singapore / SGT | ap-southeast |
| Japan / JST | ap-northeast |
| Australia / AEST / AEDT | ap-southeast |
| No signal | (omit — provision stage uses first available) |
The provision stage maps REGION_HINT + CLOUD_PROVIDER to an actual Aura region from the tenant's available configurations.
If APP_TYPE=graphrag, also collect:
EMBEDDING_PROVIDER = openai | cohere | ollama | other
EMBEDDING_MODEL = <model name, e.g. text-embedding-3-small>Special cases
DB_TARGET=existing: ask for URI, username, password, database name. Write to .env immediately, skip provision stage.
DATA_SOURCE=demo: ask which demo dataset:
- Movies — classic graph, great for beginners
- Northwind — e-commerce (orders, products, customers)
- StackOverflow — Q&A network
- Companies — corporate KG with news and embeddings
- Other — check https://github.com/neo4j-graph-examples
Detect execution method
Detect EXEC_METHOD for all subsequent Cypher execution. See ${CLAUDE_SKILL_DIR}/references/capabilities/execute-cypher.md for details on each option.
# Priority order:
# 1. MCP (neo4j-mcp running as MCP server in this session)
# 2. cypher-shell
# 3. Query API (HTTP curl — always works)
which cypher-shell 2>/dev/null && EXEC_METHOD=cypher-shell || EXEC_METHOD=query-api
# Override to mcp if neo4j-mcp tools are available in this session
echo "EXEC_METHOD=$EXEC_METHOD"Store EXEC_METHOD alongside the other variables. Reference only the relevant section of ${CLAUDE_SKILL_DIR}/references/capabilities/execute-cypher.md in subsequent stages.
On Completion — write to progress.md
### 1-context
status: done
MODE=<autonomous|hitl>
DOMAIN=<value>
USE_CASE=<value>
EXPERIENCE=<beginner|intermediate|advanced>
DB_TARGET=<value>
DATA_SOURCE=<value>
APP_TYPE=<value>
EXEC_METHOD=<mcp|cypher-shell|query-api>
CLOUD_PROVIDER=<gcp|aws|azure — omit if no preference stated>
REGION_HINT=<e.g. europe-west, us-east, sa-east — omit if no signal>Include EMBEDDING_PROVIDER and EMBEDDING_MODEL if APP_TYPE=graphrag.
Completion condition
All variables known + EXEC_METHOD determined. Summarize before proceeding:
Got it — here's your plan:
Domain: <DOMAIN>
Use-case: <USE_CASE>
Database: <DB_TARGET>
Data: <DATA_SOURCE>
Build: <APP_TYPE>
Starting now — I'll provision your database and design a data model.Stage 2 — provision
Provision a running Neo4j database and save credentials to .env.
Local DB path — handle DB_TARGET first
Check progress.md for DB_TARGET before doing anything Aura-related:
aura-freeoraura-pro→ proceed to Aura REST API section belowlocal-docker→ start Neo4j container and write.env(see below) → skip Aura sectionlocal-desktop→ ask user to start Neo4j Desktop → write.env→ skip Aura sectionexisting→ user provided credentials → write.env→ skip everything
DB_TARGET=local-docker — Docker provisioning flow
In HITL mode: ask the user for a password (or suggest password123 as default). In autonomous mode: use password123 as the default password.
# 1. Check Docker is available
docker --version 2>/dev/null || { echo "Docker not found — please install Docker Desktop"; exit 1; }
# 2. Start Neo4j — with persistent data volume so data survives container restarts
# Remove any leftover container with the same name first
PASS="password123" # change this if the user provided a different password
docker rm -f neo4j-dev 2>/dev/null || true
docker run -d \
--name neo4j-dev \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH="neo4j/${PASS}" \
-e NEO4J_ACCEPT_LICENSE_AGREEMENT=yes \
-v "$(pwd)/neo4j-data:/data" \
neo4j:enterprise
echo "✓ Container started (data persisted to $HOME/neo4j-dev/data) — waiting for Bolt..."Why `-v $(pwd)/neo4j-data:/data`? Without a volume, all data is lost when the
container is removed. With it, docker rm neo4j-dev && docker run ... keeps your graph.# 3. Wait for Neo4j to accept connections (save as /tmp/wait_neo4j.py — temp file, not in project)
from neo4j import GraphDatabase
import time, sys
for attempt in range(18): # up to 90s
try:
d = GraphDatabase.driver("bolt://localhost:7687", auth=("neo4j", "password123"))
d.verify_connectivity()
d.close()
print(f"✓ Neo4j ready after {attempt * 5}s")
sys.exit(0)
except Exception as e:
print(f" [{attempt+1}/18] not ready yet ({type(e).__name__}), retrying in 5s...")
time.sleep(5)
print("Neo4j did not start in time"); sys.exit(1).venv/bin/python3 /tmp/wait_neo4j.py# 4. Write .env
cat > .env << 'EOF'
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=password123
NEO4J_DATABASE=neo4j
EOF
echo "✓ .env written"On Completion for local-docker — write to progress.md:
### 2-provision
status: done
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_DATABASE=neo4j
CONTAINER_NAME=neo4j-devDB_TARGET=local-desktop — Neo4j Desktop flow
Tell the user:
"Please open Neo4j Desktop and start your local database. Once it's running, share the
password you set and confirm it's on the default port (bolt://localhost:7687)."
Once confirmed, write .env:
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=<password user provided>
NEO4J_DATABASE=neo4jOn Completion for local-desktop — write to progress.md:
### 2-provision
status: done
NEO4J_URI=bolt://localhost:7687
NEO4J_USERNAME=neo4j
NEO4J_DATABASE=neo4j---
Aura REST API
Use these three endpoints directly — no extra tooling needed. For other operations (list, delete, pause, resume, update), fetch the live OpenAPI spec at runtime: GET https://api.neo4j.io/openapi.json or browse https://neo4j.com/docs/aura/platform/api/specification/
Step P-1 — Collect Aura API credentials
aura.env — account-level API credentials (reusable across instances). .env — per-instance DB connection details (written later in P3). Keep separate so writing .env never overwrites the API key.
Check in order: 1. aura.env exists → load it with Python (see Step P0) → proceed 2. Environment variables CLIENT_ID / CLIENT_SECRET (or AURA_CLIENT_ID / AURA_CLIENT_SECRET) already set → proceed 3. Neither found → ask the user:
"To provision an Aura database I need your Aura API credentials.
Please go to https://console.neo4j.io → Account Settings → API credentials,
create a new client, and paste the Client ID and Client Secret here."
Once received, save to aura.env (never .env):
cat > aura.env << EOF
CLIENT_ID=<value>
CLIENT_SECRET=<value>
# Strongly recommended for users with multiple organisations or projects —
# without these the API picks the first org/project alphabetically which may be wrong.
# Find them at console.neo4j.io → your project → Settings.
# PROJECT_ID=<project/tenant id>
# ORGANIZATION_ID=<organisation id>
EOF
# aura.env is already in .gitignore from the prerequisites stageThe console generates keys named CLIENT_ID / CLIENT_SECRET. Both that form and AURA_CLIENT_ID / AURA_CLIENT_SECRET are accepted.
Steps P0–P3 — Provision via Python script
Run the entire provision flow as a single Python script. Env vars set in one Bash tool call are lost in the next — never split across multiple commands.
#!/usr/bin/env python3
"""
provision_aura.py — run this script to provision an Aura instance.
Reads aura.env, creates the instance, polls until running, writes .env.
Idempotent: exits immediately if already provisioned or in progress.
"""
import atexit, fcntl, json, os, pathlib, sys, time, urllib.request, urllib.error
# ── Idempotency guards — acquired before any network call ────────────────────
_env_path = pathlib.Path(".env")
_lock_path = pathlib.Path(".provision.lock")
# Guard 1: already done
if _env_path.exists() and "NEO4J_URI=neo4j" in _env_path.read_text():
print("✓ .env already exists with valid URI — skipping (DB already running)")
sys.exit(0)
# Guard 2: atomic exclusive file lock — immune to TOCTOU races
_lock_fh = _lock_path.open("a") # "a" so it always exists
try:
fcntl.flock(_lock_fh.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print("✓ Provisioning already in progress (lock held) — skipping")
sys.exit(0)
_lock_fh.write(f"{os.getpid()}\n"); _lock_fh.flush()
atexit.register(lambda: (_lock_path.unlink(missing_ok=True)))
print(f" Lock acquired (PID {os.getpid()})")
from dotenv import dotenv_values
# ── Load aura.env ────────────────────────────────────────────────────────────
env = dotenv_values("aura.env")
CLIENT_ID = env.get("CLIENT_ID") or env.get("AURA_CLIENT_ID")
CLIENT_SECRET = env.get("CLIENT_SECRET") or env.get("AURA_CLIENT_SECRET")
PROJECT_ID = env.get("PROJECT_ID") # optional — skip discovery if set
ORG_ID = env.get("ORGANIZATION_ID") # optional — skip discovery if set
assert CLIENT_ID and CLIENT_SECRET, "CLIENT_ID / CLIENT_SECRET missing from aura.env"
def api(method, path, token=None, body=None, base="https://api.neo4j.io"):
url = base + path
data = json.dumps(body).encode() if body else None
headers = {"Content-Type": "application/json"}
if token:
headers["Authorization"] = f"Bearer {token}"
req = urllib.request.Request(url, data=data, headers=headers, method=method)
try:
return json.loads(urllib.request.urlopen(req).read())
except urllib.error.HTTPError as e:
raise RuntimeError(f"{method} {path} → {e.code}: {e.read().decode()}") from e
# ── Token ────────────────────────────────────────────────────────────────────
# Endpoint: /oauth/token • JSON body • NOT /oauth2/token • NOT form-encoded
token = api("POST", "/oauth/token", body={
"grant_type": "client_credentials",
"client_id": CLIENT_ID,
"client_secret": CLIENT_SECRET,
})["access_token"]
print(f"✓ Token obtained ({token[:16]}...)")
# ── Resolve org + project (v2beta1 for correct scoping) ──────────────────────
if not ORG_ID:
orgs = api("GET", "/v2beta1/organizations", token)["data"]
ORG_ID = orgs[0]["id"]
print(f" Discovered ORGANIZATION_ID={ORG_ID} ({orgs[0]['name']})")
else:
print(f" Using ORGANIZATION_ID={ORG_ID} from aura.env")
if not PROJECT_ID:
projects = api("GET", f"/v2beta1/organizations/{ORG_ID}/projects", token)["data"]
PROJECT_ID = projects[0]["id"]
print(f" Discovered PROJECT_ID={PROJECT_ID} ({projects[0]['name']})")
else:
print(f" Using PROJECT_ID={PROJECT_ID} from aura.env")
# ── Read context from progress.md ─────────────────────────────────────────────
import re as _re
_progress = pathlib.Path("progress.md").read_text() if pathlib.Path("progress.md").exists() else ""
# DB_TARGET mapping: aura-free → free-db aura-pro → professional-db
# NOTE: free-db is GCP-only (3 regions: europe-west1, us-central1, asia-southeast1).
# professional-db supports GCP (12), AWS (10), Azure (8).
_db_target_m = _re.search(r"^DB_TARGET=(\S+)", _progress, _re.MULTILINE)
DB_TYPE = "professional-db" if _db_target_m and "pro" in _db_target_m.group(1) else "free-db"
print(f" DB_TYPE={DB_TYPE} (from DB_TARGET in progress.md)")
tenant_data = api("GET", f"/v1beta5/tenants/{PROJECT_ID}", token)
# Each config: {"type", "cloud_provider", "region", "region_name", "memory", "storage"}
# region_name is human-readable, e.g. "Belgium (europe-west1)", "US East, N. Virginia (us-east-1)"
# The same (cloud_provider, region) pair appears multiple times — once per memory size.
# Deduplicate immediately using a dict keyed by (cloud_provider, region); first entry wins.
unique_configs = list({
(c["cloud_provider"], c["region"]): c
for c in tenant_data["data"].get("instance_configurations", [])
if c["type"] == DB_TYPE
}.values())
print(f" Available {DB_TYPE} regions ({len(unique_configs)}): "
f"{[(c['cloud_provider'], c['region_name']) for c in unique_configs]}")
# ── Pick best region ───────────────────────────────────────────────────────────
# CLOUD_PROVIDER = user's explicit preference (from context stage, aura-pro only)
# REGION_HINT = geographic area inferred from signals (never asked)
# (_re and _progress already defined above)
_cp = _re.search(r"^CLOUD_PROVIDER=(\S+)", _progress, _re.MULTILINE)
_rh = _re.search(r"^REGION_HINT=(\S+)", _progress, _re.MULTILINE)
PREF_PROVIDER = _cp.group(1).lower() if _cp else None # e.g. "aws"
REGION_HINT = _rh.group(1).lower() if _rh else None # e.g. "europe-west"
# REGION_HINT → actual Aura region identifiers (from real tenant API data)
REGION_KEYWORDS = {
"europe-west": ["europe-west1", "europe-west2", "eu-west-1", "eu-west-3",
"francecentral", "uksouth", "westeurope"],
"europe-east": ["europe-west3", "eu-central-1"],
"us-east": ["us-east-1", "us-east1", "us-east-2", "eastus"],
"us-west": ["us-west-2", "us-west1", "westus3"],
"us-central": ["us-central1"],
"sa-east": ["sa-east-1", "brazilsouth"],
"ap-southeast": ["ap-southeast-1", "ap-southeast-2", "asia-southeast1",
"australia-southeast1"],
"ap-northeast": ["asia-east1", "asia-east2", "koreacentral"],
"ap-south": ["ap-south-1", "asia-south1", "centralindia"],
}
keywords = REGION_KEYWORDS.get(REGION_HINT, []) if REGION_HINT else []
# Cheapest/lowest-latency anchor per provider (used when no hint matches)
PROVIDER_DEFAULTS = {
"gcp": "europe-west1", # Belgium — original GCP region, consistently cheapest
"aws": "us-east-1", # N. Virginia — cheapest AWS, most services available
"azure": "eastus", # Virginia — cheapest Azure
}
CLOUD_PROVIDER, REGION, REGION_NAME = None, None, ""
def _pick(pool, provider=None, kw_list=None):
candidates = [c for c in pool if not provider or c["cloud_provider"] == provider]
if kw_list:
for kw in kw_list:
for c in candidates:
if kw in c["region"]:
return c
return None
return candidates[0] if candidates else None
# [1] User's preferred provider + closest region to geo hint
if PREF_PROVIDER and keywords:
c = _pick(unique_configs, PREF_PROVIDER, keywords)
if c:
CLOUD_PROVIDER, REGION, REGION_NAME = c["cloud_provider"], c["region"], c.get("region_name","")
print(f" [1] provider pref + geo: {CLOUD_PROVIDER}/{REGION} ({REGION_NAME})")
# [2] User's preferred provider + cheapest default region for that provider
if not CLOUD_PROVIDER and PREF_PROVIDER:
c = _pick(unique_configs, PREF_PROVIDER, [PROVIDER_DEFAULTS.get(PREF_PROVIDER, "")])
if not c:
c = _pick(unique_configs, PREF_PROVIDER) # any region in that provider
if c:
CLOUD_PROVIDER, REGION, REGION_NAME = c["cloud_provider"], c["region"], c.get("region_name","")
print(f" [2] provider pref, default region: {CLOUD_PROVIDER}/{REGION} ({REGION_NAME})")
# [3] No provider preference — geo hint across all providers
if not CLOUD_PROVIDER and keywords:
c = _pick(unique_configs, kw_list=keywords)
if c:
CLOUD_PROVIDER, REGION, REGION_NAME = c["cloud_provider"], c["region"], c.get("region_name","")
print(f" [3] geo hint (any provider): {CLOUD_PROVIDER}/{REGION} ({REGION_NAME})")
# [4] No hint at all — cheapest defaults (GCP europe-west1, then AWS us-east-1, then Azure eastus)
if not CLOUD_PROVIDER:
for provider, default_region in PROVIDER_DEFAULTS.items():
c = _pick(unique_configs, provider, [default_region])
if c:
CLOUD_PROVIDER, REGION, REGION_NAME = c["cloud_provider"], c["region"], c.get("region_name","")
print(f" [4] cheapest default: {CLOUD_PROVIDER}/{REGION} ({REGION_NAME})")
break
# [5] Absolute fallback
if not CLOUD_PROVIDER:
if unique_configs:
c = unique_configs[0]
CLOUD_PROVIDER, REGION, REGION_NAME = c["cloud_provider"], c["region"], c.get("region_name","")
print(f" [5] first available: {CLOUD_PROVIDER}/{REGION}")
else:
CLOUD_PROVIDER, REGION = "gcp", "europe-west1"
print(f" [5] hardcoded fallback (no tenant configs): {CLOUD_PROVIDER}/{REGION}")
# ── Create instance ────────────────────────────────────────────────────────────
try:
result = api("POST", "/v1beta5/instances", token, body={
"name": "myapp-db",
"tenant_id": PROJECT_ID,
"cloud_provider": CLOUD_PROVIDER,
"region": REGION,
"type": DB_TYPE,
"memory": "1GB",
})["data"]
INSTANCE_ID = result["id"]
PASSWORD = result["password"] # shown only once — captured here
print(f"✓ Instance created: {INSTANCE_ID} ({CLOUD_PROVIDER}/{REGION})")
except RuntimeError as e:
if "quota" not in str(e).lower() and "limit" not in str(e).lower():
raise
# Free quota exceeded — fall back to a Pro trial instance (also free for new accounts)
print(f" Free quota exceeded. Falling back to professional-db trial instance...")
pro_configs = [c for c in available if c["type"] == "professional-db"]
if pro_configs:
CLOUD_PROVIDER = pro_configs[0]["cloud_provider"]
REGION = pro_configs[0]["region"]
result = api("POST", "/v1beta5/instances", token, body={
"name": "myapp-db",
"tenant_id": PROJECT_ID,
"cloud_provider": CLOUD_PROVIDER,
"region": REGION,
"type": "professional-db",
"memory": "1GB",
})["data"]
INSTANCE_ID = result["id"]
PASSWORD = result["password"]
print(f"✓ Pro trial instance created: {INSTANCE_ID}")
# ── Poll until running ────────────────────────────────────────────────────────
CONNECTION = ""
for i in range(1, 25):
status_data = api("GET", f"/v1beta5/instances/{INSTANCE_ID}", token)["data"]
status = status_data.get("status", "")
CONNECTION = status_data.get("connection_url", "")
print(f" [{i}/24] {status}")
if status == "running":
break
time.sleep(15)
else:
raise RuntimeError("Instance did not reach 'running' after 6 minutes")
# ── Wait for Bolt to accept connections ──────────────────────────────────────
# Aura reports "running" before the Bolt port is actually ready.
# Verify connectivity with retries before writing .env.
try:
from neo4j import GraphDatabase as _GDB
_connected = False
for _attempt in range(12): # up to 2 min
try:
_d = _GDB.driver(CONNECTION, auth=("neo4j", PASSWORD))
_d.verify_connectivity()
_d.close()
_connected = True
print(f" Bolt ready after {(_attempt) * 10}s")
break
except Exception as _e:
print(f" [{_attempt+1}/12] Bolt not ready yet ({type(_e).__name__}), waiting 10s...")
time.sleep(10)
if not _connected:
raise RuntimeError("Bolt port never became ready after 2 minutes")
except ImportError:
# neo4j driver not installed yet — skip connectivity check, add a 30s safety buffer
print(" neo4j driver not available for connectivity check — sleeping 30s as safety buffer")
time.sleep(30)
# ── Write .env ────────────────────────────────────────────────────────────────
pathlib.Path(".env").write_text(
f"NEO4J_URI={CONNECTION}\n"
f"NEO4J_USERNAME=neo4j\n"
f"NEO4J_PASSWORD={PASSWORD}\n"
f"NEO4J_DATABASE=neo4j\n"
)
print(f"✓ .env written URI={CONNECTION}")
print(f"✓ DONE — instance {INSTANCE_ID} is running")Write to scripts/provision_aura.py. Do not run directly — use the recommended flow below with background launch + log file.
---
Aura CLI Quick Reference
Installation
# macOS
brew install neo4j/tap/aura-cli # if homebrew tap exists
# or: download binary from https://github.com/neo4j/aura-cli/releases/latest
sudo mv aura-cli /usr/local/bin/ && chmod +x /usr/local/bin/aura-cli
# Verify
aura-cli --versionCredential setup
# Generate Client ID + Secret at https://console.neo4j.io → Account Settings → API credentials
aura-cli credential add \
--name "default" \
--client-id $AURA_CLIENT_ID \
--client-secret $AURA_CLIENT_SECRETInstance lifecycle
# Create Free instance (512MB, GCP)
aura-cli instance create \
--name "myapp-db" \
--cloud-provider gcp \
--region europe-west1 \
--type free-db \
--output json
# Create Pro instance (1GB, AWS)
aura-cli instance create \
--name "myapp-prod" \
--cloud-provider aws \
--region us-east-1 \
--type professional-db \
--memory 1 \
--output json
# List instances
aura-cli instance list --output json
# Get single instance status (check for "running")
aura-cli instance get <INSTANCE_ID> --output json
# Pause / resume (cost saving)
aura-cli instance pause <INSTANCE_ID>
aura-cli instance resume <INSTANCE_ID>
# Delete
aura-cli instance delete <INSTANCE_ID>Regions by cloud provider
| Provider | Available Regions |
|---|---|
| GCP | us-central1, us-east1, europe-west1, europe-west3, asia-east1, asia-southeast1 |
| AWS | us-east-1, us-west-2, eu-west-1, eu-central-1, ap-southeast-1 |
| Azure | eastus, westeurope, southeastasia |
Poll for running status (bash)
INSTANCE_ID="<id>"
for i in $(seq 1 24); do
STATUS=$(aura-cli instance get $INSTANCE_ID --output json | \
python3 -c "import sys,json; d=json.load(sys.stdin); print(d.get('status','unknown'))")
echo "[$i/24] Status: $STATUS"
[ "$STATUS" = "running" ] && { echo "Instance ready"; break; }
sleep 15
done---
Docker Quick Reference
# Basic (ephemeral — data lost on container remove)
docker run -d \
--name neo4j-dev \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password123 \
-e NEO4J_ACCEPT_LICENSE_AGREEMENT=yes \
neo4j:enterprise
# Recommended — persistent data volume
docker run -d \
--name neo4j-dev \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password123 \
-e NEO4J_ACCEPT_LICENSE_AGREEMENT=yes \
-v $(pwd)/neo4j-data:/data \
neo4j:enterprise
# With plugins (APOC + GDS)
docker run -d \
--name neo4j-dev \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password123 \
-e NEO4J_ACCEPT_LICENSE_AGREEMENT=yes \
-e NEO4J_PLUGINS='["apoc","graph-data-science"]' \
neo4j:enterprise
# Persistent data volume
docker run -d \
--name neo4j-dev \
-p 7474:7474 -p 7687:7687 \
-e NEO4J_AUTH=neo4j/password123 \
-e NEO4J_ACCEPT_LICENSE_AGREEMENT=yes \
-v $HOME/neo4j/data:/data \
neo4j:enterprise
# Check logs
docker logs neo4j-dev -f
# Stop / remove
docker stop neo4j-dev && docker rm neo4j-dev---
Connectivity Verification
cypher-shell
cypher-shell -a "neo4j+s://xxxxx.databases.neo4j.io" \
-u neo4j -p "<password>" \
"RETURN 'connected' AS status"Python
from neo4j import GraphDatabase
driver = GraphDatabase.driver(
"neo4j+s://xxxxx.databases.neo4j.io",
auth=("neo4j", "<password>")
)
driver.verify_connectivity()
print("Connected")
driver.close()Node.js
const neo4j = require('neo4j-driver');
const driver = neo4j.driver(
'neo4j+s://xxxxx.databases.neo4j.io',
neo4j.auth.basic('neo4j', '<password>')
);
await driver.verifyConnectivity();
console.log('Connected');
await driver.close();---
Neo4j Query API (HTTP — no driver required)
Useful for connectivity checks and scripting when no driver is installed:
# Aura: host is the bolt URI without the scheme
HOST="xxxxx.databases.neo4j.io"
curl -s -X POST "https://${HOST}/db/neo4j/query/v2" \
-H "Content-Type: application/json" \
-u "neo4j:<password>" \
-d '{"statement": "MATCH (n) RETURN count(n) AS total"}' \
| python3 -c "import sys,json; d=json.load(sys.stdin); print(d)"
# Local Docker
curl -s -X POST "http://localhost:7474/db/neo4j/query/v2" \
-H "Content-Type: application/json" \
-u "neo4j:password123" \
-d '{"statement": "RETURN 1"}'---
URI Schemes
| Scheme | Use case |
|---|---|
neo4j+s:// | Aura (TLS required) |
bolt+s:// | Self-hosted with TLS |
bolt:// | Local development (no TLS) |
neo4j:// | Cluster routing, no TLS |
---
Parallelise with offline work (saves 2–3 min)
Aura provisioning takes 2–4 minutes. Everything that doesn't touch the database can run during that wait.
What can be done before the DB is running (no connection needed):
- Stage 3: design the model, write
schema/schema.json+schema/schema.cypher - Stage 4: write
data/generate.py, run it (pure Python → CSVs), writedata/import.py - Stage 6: write
queries/queries.cypher(text only — validation runs later)
What must wait for the DB:
- Apply
schema/schema.cypher(constraints + indexes) - Run
data/import.py(loads CSVs into Neo4j) - Run
validate_queries.py(executes queries against live DB) - Stage 5: generate browser URL (needs
NEO4J_URIfrom.env)
Recommended flow after writing provision_aura.py:
# Step 1 — launch provision in background with unbuffered output (never run twice, never kill)
mkdir -p scripts
PYTHONUNBUFFERED=1 python3 scripts/provision_aura.py > scripts/provision.log 2>&1 &
echo "Provision PID=$!"
sleep 3 && head -10 scripts/provision.log # confirm it started
# Step 2 — do all offline work while DB spins up (no DB needed):
# - read 3-model.md → design model → write schema/schema.json + schema/schema.cypher
# - read 4-load.md → write data/import.py
# - read 6-query.md → write queries/queries.cypher
# Step 3 — wait for .env to appear (provision script writes it when DB is running)
until grep -q "NEO4J_URI=neo4j" .env 2>/dev/null; do sleep 10; echo "waiting for DB..."; done
echo "✓ DB ready" && tail -5 scripts/provision.logIMPORTANT — never kill or re-run the provision process:
PYTHONUNBUFFERED=1ensures output appears immediately in the log- If the log appears empty after 5s, check
ps aux | grep provision— if the process is running, it is working; just wait - Never `kill` a running provision process — this releases the lock and allows a duplicate to start
- If
.envalready exists with a valid URI, skip provision entirely — the DB is already running
If offline-written files need fixing after DB validation, only execution errors need addressing — structure is already correct.
On Completion — write to progress.md
Write the provision script to scripts/provision_aura.py (not the project root).
### 2-provision
status: done
NEO4J_URI=<value from .env>
NEO4J_USERNAME=<value from .env>
NEO4J_DATABASE=<value from .env, usually "neo4j">
INSTANCE_ID=<Aura instance ID, e.g. "f1cad593" — needed for cleanup>
files=scripts/provision_aura.py.env File Template
NEO4J_URI=neo4j+s://xxxxx.databases.neo4j.io
NEO4J_USERNAME=neo4j
NEO4J_PASSWORD=<generated-password>
NEO4J_DATABASE=neo4jStage 3 — model
Design or discover the graph data model for the use-case.
Autonomous mode check — do this FIRST
Check progress.md for MODE=autonomous first.
If `MODE=autonomous`: design the model without any review pause. Do not show a "does this look right?" message. Do not wait for confirmation at any point. Auto-approve your design and proceed immediately through all steps M4→M5→completion.
If `MODE=hitl`: Step M3 (model review) is active — show the draft and wait for user approval.
Path selection
DATA_SOURCE=demo → use the demo dataset's pre-built schema (see domain-patterns.md)
skip to load stage directly
DB_TARGET=existing → introspect existing schema (Path D below)
DATA_SOURCE=csv → inspect CSV headers first, derive model (Path B)
DATA_SOURCE=documents → use KG/RAG model template from domain-patterns.md
otherwise → greenfield modeling (Path C)Path A — Demo dataset schema
Look up the schema in ${CLAUDE_SKILL_DIR}/references/domain-patterns.md for the chosen demo. Write schema.json from that template. Write schema.cypher with DDL. Proceed to load.
Path B — CSV-first modeling
Inspect headers and sample rows before designing anything:
for f in ./data/*.csv; do
echo "=== $f ==="
head -4 "$f"
echo
doneDerive model from structure:
- Each entity-centric file → candidate node label (look for
_idcolumns as primary key) - Foreign key column in one file referencing another → relationship
- Normalize names:
customer_idcolumn →Customernode withidproperty - Numeric columns → properties; date columns → datetime properties
Path C — Greenfield modeling
Use ${CLAUDE_SKILL_DIR}/references/domain-patterns.md as starting templates. Adapt to the user's use-case.
Principles:
- 3–6 node labels (more for advanced users)
- 3–8 relationship types with clear direction and semantics
- One natural primary key per node (enables MERGE safety)
- 2–5 properties per node with realistic types
- At least one property suitable for fulltext search (e.g.
name,title,description)
Path D — Existing DB introspection
# Via neo4j-mcp get-schema tool (preferred)
# OR via cypher-shell:
source .env
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CALL db.schema.visualization() YIELD nodes, relationships RETURN nodes, relationships"Write discovered schema to schema.json. Skip to query stage.
Step M3 — Model review (HITL only)
AUTONOMOUS MODE (all context provided upfront): SKIP THIS STEP ENTIRELY. Do not show the model for review. Do not ask "does this look right?". Proceed directly to Step M4.
HITL only — show the proposed model and wait for approval:
Here's the graph model I'm proposing for <USE_CASE>:
Nodes:
- <Label> {<primaryKey>, <prop1>, <prop2>}
...
Relationships:
- (<LabelA>)-[:<REL_TYPE>]->(<LabelB>)
...
Does this look right? Anything to add, rename, or remove?
(Reply "ok" to proceed, or describe changes.)Step M4 — Write schema/schema.json
mkdir -p schema{
"nodes": [
{"label": "Person", "primaryKey": "id", "properties": ["id","name","email","createdAt"]}
],
"relationships": [
{"type": "FOLLOWS", "from": "Person", "to": "Person", "properties": []}
]
}Write to schema/schema.json.
Step M5 — Write schema/schema.cypher (DDL)
Write to schema/schema.cypher.
Constraints before indexes. Constraints before data.
CYPHER 25
// Uniqueness constraints — required for fast MERGE
CREATE CONSTRAINT person_id IF NOT EXISTS FOR (p:Person) REQUIRE p.id IS UNIQUE;
// Lookup indexes for common query patterns
CREATE INDEX person_name IF NOT EXISTS FOR (p:Person) ON (p.name);For vector/graphrag use-cases, add the vector index here too:
CYPHER 25
CREATE VECTOR INDEX chunk_embeddings IF NOT EXISTS
FOR (c:Chunk) ON (c.embedding)
OPTIONS { indexConfig: { `vector.dimensions`: 1536, `vector.similarity_function`: 'cosine' } };On Completion — write to progress.md
### 3-model
status: done
labels=<comma-separated node labels>
relationships=<comma-separated relationship types>
constraints=<number applied>
files=schema/schema.json,schema/schema.cypher
sample_id=<a real primary-key value from the data, e.g. "p1" — used for query params>Completion condition
schema/schema.jsonwritten with at least 2 nodes and 1 relationshipschema/schema.cypherwritten with at least 1 uniqueness constraint- HITL: user approved; Autonomous: auto-approved (no pause)
Stage 4 — load
Import data into the database. Always apply schema constraints first.
Step L0 — Apply schema constraints (always, before any data)
source .env
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" --file schema/schema.cypherOr via Python if cypher-shell unavailable:
from neo4j import GraphDatabase; import os
driver = GraphDatabase.driver(os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"]))
with driver.session() as s:
for stmt in open("schema/schema.cypher").read().split(";"):
stmt = stmt.strip()
if stmt and not stmt.startswith("//"):
s.run(stmt)
driver.close()
print("Schema applied")Import rules (apply to all paths)
- MERGE nodes first — complete all node MERGE statements before any relationship MERGE
- MERGE relationships second — only after all endpoint node types are loaded
- Batch size: 500 rows per call — pass as
$rowsparameter list from Python - Use MERGE not CREATE — idempotent, safe to re-run
- All scripts go in `data/` — user can re-run them for updates
Preferred pattern — Python batch loading via DataFrame
Load into pandas DataFrame, push to Neo4j in batches via driver.execute_query(..., rows=batch). Works with any source (local files, HTTPS, S3/GCS, Parquet, relational DBs, MongoDB) — no Neo4j import directory access required (works on Aura).
import os
import pandas as pd
from neo4j import GraphDatabase
from dotenv import load_dotenv
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
BATCH = 500
def load_batches(query: str, rows: list[dict]) -> int:
total = 0
for i in range(0, len(rows), BATCH):
records, summary, _ = driver.execute_query(query, rows=rows[i:i+BATCH])
total += summary.counters.nodes_created + summary.counters.relationships_created
return total
# ── Step 0: apply constraints before any data (idempotent) ────────────────────
schema = open("schema/schema.cypher").read()
with driver.session() as s:
for stmt in schema.split(";"):
stmt = stmt.strip()
if stmt and not stmt.startswith("//"):
s.run(stmt)
print("✓ Constraints applied")
# ── Read data — swap in any pandas-compatible source ──────────────────────────
# CSV (local or HTTPS): pd.read_csv("https://data.neo4j.com/northwind/products.csv")
# Parquet / S3: pd.read_parquet("s3://bucket/file.parquet")
# Relational (SQLAlchemy):pd.read_sql("SELECT * FROM products", engine)
# MongoDB: pd.DataFrame(collection.find())
products = pd.read_csv("https://data.neo4j.com/northwind/products.csv")
categories = pd.read_csv("https://data.neo4j.com/northwind/categories.csv")
# ── Phase 1: all node types (MERGE nodes before relationships) ─────────────────
n = load_batches("""
UNWIND $rows AS row
MERGE (p:Product {productID: row.productID})
SET p.productName = row.productName,
p.unitPrice = toFloat(row.unitPrice),
p.unitsInStock = toInteger(row.unitsInStock)
""", products.to_dict("records"))
print(f"Products: {n}")
n = load_batches("""
UNWIND $rows AS row
MERGE (c:Category {categoryID: row.categoryID})
SET c.categoryName = row.categoryName,
c.description = row.description
""", categories.to_dict("records"))
print(f"Categories: {n}")
# ── Phase 2: relationships (after all nodes exist) ─────────────────────────────
n = load_batches("""
UNWIND $rows AS row
MATCH (p:Product {productID: row.productID})
MATCH (c:Category {categoryID: row.categoryID})
MERGE (p)-[:PART_OF]->(c)
""", products.to_dict("records"))
print(f"PART_OF rels: {n}")
driver.close()Run: .venv/bin/python3 data/import.py
Type coercion in Cypher vs Python
Prefer coercing types in Python before passing rows (faster, avoids Cypher toFloat()):
products["unitPrice"] = pd.to_numeric(products["unitPrice"], errors="coerce")
products["unitsInStock"] = pd.to_numeric(products["unitsInStock"], errors="coerce").astype("Int64")Then use row.unitPrice directly in Cypher without wrapping functions.
---
Path A — Demo dataset
source .env
# Movies
curl -s https://raw.githubusercontent.com/neo4j-graph-examples/movies/main/data/movies.cypher \
| cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD"For other demos, fetch the import URL from https://github.com/neo4j-graph-examples.
Path B — Synthetic data
Two-step approach: generate CSVs first, then import via DataFrame. This is faster to write, easier to inspect, and reuses the same batch loading pattern as Path C.
Step B1 — Generate CSVs (data/generate.py)
Generate one CSV per entity type using Python's csv module — no DB connection needed:
import csv, os, random
from datetime import datetime, timedelta
os.makedirs("data", exist_ok=True)
random.seed(42)
CITIES = ["London", "New York", "Berlin", "Tokyo", "Sydney"]
# ── Nodes ─────────────────────────────────────────────────────────────────────
with open("data/persons.csv", "w", newline="") as f:
w = csv.DictWriter(f, ["id", "name", "age", "city", "joined_at"])
w.writeheader()
for i in range(1, 201):
w.writerow({
"id": f"p{i}",
"name": f"Person {i}",
"age": random.randint(18, 65),
"city": random.choice(CITIES),
"joined_at": (datetime.now() - timedelta(days=random.randint(0, 730))).strftime("%Y-%m-%d"),
})
# (add more node CSVs for other labels in schema.json)
# ── Relationships ─────────────────────────────────────────────────────────────
ids = [f"p{i}" for i in range(1, 201)]
with open("data/follows.csv", "w", newline="") as f:
w = csv.DictWriter(f, ["from_id", "to_id"])
w.writeheader()
for src in ids:
for tgt in random.sample(ids, k=random.randint(3, 15)):
if src != tgt:
w.writerow({"from_id": src, "to_id": tgt})
print("✓ CSVs written:", [f for f in os.listdir("data") if f.endswith(".csv")])Run: .venv/bin/python3 data/generate.py
Step B2 — Import CSVs (data/import.py)
Use the standard batch loading pattern — same as Path C:
import os, pandas as pd
from neo4j import GraphDatabase
from dotenv import load_dotenv
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
BATCH = 500
def load_batches(query, rows):
total = 0
for i in range(0, len(rows), BATCH):
_, summary, _ = driver.execute_query(query, rows=rows[i:i+BATCH])
total += summary.counters.nodes_created + summary.counters.relationships_created
return total
# ── Phase 1: nodes ─────────────────────────────────────────────────────────────
persons = pd.read_csv("data/persons.csv")
n = load_batches("""
UNWIND $rows AS row
MERGE (p:Person {id: row.id})
SET p.name = row.name, p.age = toInteger(row.age),
p.city = row.city, p.joinedAt = date(row.joined_at)
""", persons.to_dict("records"))
print(f" Person nodes: {n}")
# ── Phase 2: relationships ──────────────────────────────────────────────────────
follows = pd.read_csv("data/follows.csv")
n = load_batches("""
UNWIND $rows AS row
MATCH (a:Person {id: row.from_id})
MATCH (b:Person {id: row.to_id})
MERGE (a)-[:FOLLOWS]->(b)
""", follows.to_dict("records"))
print(f" FOLLOWS rels: {n}")
records, _, _ = driver.execute_query("MATCH (n) RETURN labels(n)[0] AS l, count(n) AS c")
for r in records:
print(f" {r['l']}: {r['c']}")
driver.close()Run: .venv/bin/python3 data/import.py
Path C — CSV / tabular data (any source)
Use the Python batch loading pattern above. Install dependencies first:
.venv/bin/pip install neo4j-rust-ext pandas python-dotenvAdapt the DataFrame source to match:
| Source | pandas call |
|---|---|
| Local CSV | pd.read_csv("./data/file.csv") |
| HTTPS CSV | pd.read_csv("https://…/file.csv") |
| Parquet / S3 | pd.read_parquet("s3://bucket/file.parquet") |
| PostgreSQL | pd.read_sql("SELECT * FROM table", engine) |
| MongoDB | pd.DataFrame(collection.find({}, {"_id": 0})) |
| Excel | pd.read_excel("file.xlsx") |
Always follow Phase 1 (all nodes) → Phase 2 (all relationships) regardless of source.
Path D — Document / GraphRAG pipeline (DATA_SOURCE=documents)
STOP — do NOT generate synthetic data for this path. Ingest what is already in data/. Fall back to synthetic only if the user explicitly confirms they have no files.
Step D0 — Inventory data/
find data/ -type f | sort
wc -l data/* # rough size check- Files present → proceed with ingestion of those files
- data/ empty or missing → stop and ask the user where their documents are;
offer to generate synthetic only if they confirm they have no real files
- Large files (>500 KB) → note they will take longer to embed; proceed anyway
- PDFs → set
from_pdf=Truein SimpleKGPipeline and installpypdf - Encoding issues → read with
errors="replace"and log any decoding problems
Step D1 — Install dependencies
.venv/bin/pip install neo4j-rust-ext "neo4j-graphrag[openai]>=1.13.0" python-dotenv --quiet
# If PDFs are present:
# .venv/bin/pip install pypdf --quietStep D2 — Verify .env has LLM and embedding keys
grep -E "OPENAI_API_KEY|EMBEDDING_MODEL|LLM_MODEL" .env || echo "⚠ Missing LLM keys in .env"If OPENAI_API_KEY is missing: check aura.env and append to .env.
Step D3 — Write and run import/ingest_docs.py
Follow ${CLAUDE_SKILL_DIR}/references/capabilities/kg-from-documents.md for the full pipeline template. Key points:
from neo4j_graphrag.experimental.pipeline.kg_builder import SimpleKGPipeline
from neo4j_graphrag.indexes import create_vector_index
pipeline = SimpleKGPipeline(
llm=llm,
driver=driver,
embedder=embedder,
from_pdf=False, # set True if .pdf files present
neo4j_database=os.environ.get("NEO4J_DATABASE", "neo4j"),
schema={
"node_types": NODE_TYPES, # adapt to domain
"relationship_types": RELATIONSHIP_TYPES,
"patterns": PATTERNS,
},
perform_entity_resolution=True,
)
# Load ALL files found in data/ — .txt, .md, .pdf
for path in sorted(Path("data/").glob("**/*")):
if path.is_file() and path.suffix in (".txt", ".md", ".pdf"):
try:
text = path.read_text(encoding="utf-8", errors="replace") if path.suffix != ".pdf" else None
if path.suffix == ".pdf":
await pipeline.run_async(file_path=str(path))
else:
await pipeline.run_async(text=text)
print(f" ✓ {path.name}")
except Exception as e:
print(f" ✗ {path.name}: {e}") # log and continue — don't abort
# IMPORTANT: SimpleKGPipeline does NOT create the vector index automatically.
# Always create it explicitly after ingestion using the graphrag library helper.
create_vector_index(
driver,
name="chunk_embeddings",
label="Chunk",
embedding_property="embedding",
dimensions=int(os.environ.get("EMBEDDING_DIMENSIONS", "1536")),
similarity_fn="cosine",
neo4j_database=os.environ.get("NEO4J_DATABASE", "neo4j"),
)
print(" ✓ Vector index 'chunk_embeddings' ready")Messy data — common patterns
| Problem | Symptom | Fix |
|---|---|---|
| Mixed encoding | UnicodeDecodeError | open(path, errors="replace") |
| Very large files | Slow ingestion | OK — pipeline chunks internally; just wait |
| Scanned PDFs (no text layer) | Empty chunks | Run OCR first (e.g. pytesseract) |
| Files with boilerplate headers | Low-value entity extraction | Strip headers before passing text= |
| No useful structure | LLM extracts nothing | Try schema=None — lets LLM infer types freely |
Run ingestion synchronously — do NOT use `&` or background execution. Script must complete and print "✓ Ingestion complete" before the next stage begins. LLM-based extraction is slow (minutes for large corpora) — expected, just wait.
Always inspect the actual graph schema after ingestion before writing queries:
CYPHER 25
CALL db.schema.visualization()CYPHER 25
MATCH (n) RETURN labels(n)[0] AS label, count(n) AS cnt ORDER BY cnt DESCCYPHER 25
MATCH ()-[r]->() RETURN type(r) AS rel, count(r) AS cnt ORDER BY cnt DESCSimpleKGPipeline entity label note: Extracted entities are stored under __KGBuilder__ label (not Entity, Party, etc.). The relationship from entity to its source chunk is FROM_CHUNK (not MENTIONS). The relationship from chunk to document is FROM_DOCUMENT (not HAS_CHUNK).
Use these actual labels when writing the retrieval_query for VectorCypherRetriever:
retrieval_query = """
OPTIONAL MATCH (entity:__KGBuilder__)-[:FROM_CHUNK]->(node)
RETURN node.text AS chunk_text,
collect(DISTINCT entity.name)[..5] AS entities,
score
ORDER BY score DESC
"""Step L5 — Post-import search indexes
CYPHER 25
CREATE FULLTEXT INDEX <label>_name IF NOT EXISTS
FOR (n:<Label>) ON EACH [n.name];Step L6 — Write schema/reset.cypher (always)
cat > schema/reset.cypher << 'EOF'
// Delete all data — keeps schema (constraints + indexes)
CYPHER 25
MATCH (n) CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 1000 ROWS;
EOF
echo "Reset script: cypher-shell ... --file schema/reset.cypher"Step L7 — HITL data preview pause
Show row counts and a sample before proceeding:
source .env
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CYPHER 25 MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count ORDER BY count DESC"In autonomous mode: log counts and continue.
Completion condition
MATCH (n) RETURN count(n)≥ 50- Each node label has ≥1 node
data/directory contains the import script(s) usedschema/reset.cypherexists
On Completion — write to progress.md
Record node counts per label and total relationships. For sample_id, read the first two rows of the primary node CSV:
import csv
with open("data/persons.csv") as f: # replace with your primary node CSV
row = next(csv.DictReader(f))
print(row["id"]) # use the primaryKey field from schema.jsonOr query the DB: MATCH (n:Person) RETURN n.id LIMIT 1
### 4-load
status: done
nodes=<e.g. "200 Person, 50 Post">
relationships=<e.g. "1400 FOLLOWS, 300 POSTED">
files=data/generate.py,data/import.py,schema/reset.cypher
sample_id=<actual value from first data row, e.g. "p1" or "42" or "abc-uuid">Error recovery
- Import partially failed → run
reset.cypher, re-applyschema.cypher, retry from scratch - MERGE slow → check constraint was created before import (Step L0)
- DataFrame empty → verify source URL/path and column names match schema.json
Stage 5 — explore
Deliver a visual view of the graph — the "it clicks" moment. Hard success gate. Do not skip.
Option A — Neo4j Browser standalone (always available, zero install)
Generate and print the URL:
import os, urllib.parse
from dotenv import load_dotenv
load_dotenv()
uri = os.environ.get("NEO4J_URI", "")
user = os.environ.get("NEO4J_USERNAME", "neo4j")
# Strip scheme prefix to get host
host = uri
for prefix in ["neo4j+s://", "neo4j://", "bolt+s://", "bolt://"]:
host = host.replace(prefix, "")
# Encode the connectURL parameter
connect_url = f"neo4j+s://{user}@{host}"
encoded = urllib.parse.quote(connect_url, safe="")
browser_url = f"https://browser.neo4j.io/?connectURL={encoded}"
print(f"\n🔍 See your graph:")
print(f" {browser_url}")
print(f"\n Run this query after connecting:")
print(f" MATCH (n)-[r]->(m) RETURN n,r,m LIMIT 50")Tell the user: "Open the URL above, connect with the password from .env, then run the query to see your data as a graph."
Option B — Notebook visualization (for APP_TYPE=notebook)
Use neo4j-viz — the official Neo4j Python graph visualization library:
# %pip install -q neo4j-viz
from neo4j_viz.neo4j import from_neo4j
from neo4j import GraphDatabase, RoutingControl
import os
from dotenv import load_dotenv
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
result = driver.execute_query(
"CYPHER 25 MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 50",
routing_=RoutingControl.READ
)
vg = from_neo4j(result)
vg.color_nodes(field="caption")
vg.render()Option C — VS Code Neo4j extension
Tell the user: "In VS Code, install the Neo4j extension — connects to your DB and renders Cypher results as an interactive graph directly in the editor."
Sample visualization queries by domain
Include one adapted to the user's schema:
// Social: show follow network
CYPHER 25
MATCH (p:Person)-[:FOLLOWS]->(friend:Person)
RETURN p, friend LIMIT 50;
// E-commerce: show customer orders and products
CYPHER 25
MATCH (c:Customer)-[:PLACED]->(o:Order)-[:CONTAINS]->(p:Product)
RETURN c, o, p LIMIT 50;
// Finance: show transaction network
CYPHER 25
MATCH (a:Account)-[t:TRANSFERRED_TO]->(b:Account)
RETURN a, t, b LIMIT 50;
// KG/RAG: show document-chunk-entity structure (SimpleKGPipeline)
CYPHER 25
MATCH (e:__KGBuilder__)-[:FROM_CHUNK]->(c:Chunk)-[:FROM_DOCUMENT]->(d)
RETURN e, c, d LIMIT 50;On Completion — write to progress.md
CRITICAL: This write is a hard gate. Do it immediately after generating the URL — before moving to stage 6.
### 5-explore
status: done
browser_url=<the generated https://browser.neo4j.io/... URL>
viz_method=<browser|notebook-neo4j-viz|vscode>The browser_url line in the ### 5-explore section is validated by the test harness. Write it to progress.md even if APP_TYPE=streamlit or APP_TYPE=notebook — the browser URL is always useful.
Completion condition
- Browser URL printed to stdout (Option A) AND written to progress.md, OR
- Notebook visualization cell added (Option B) AND browser_url written to progress.md, OR
- VS Code extension suggested (Option C) AND browser_url written to progress.md
Option A (browser_url) must always be delivered and recorded, regardless of APP_TYPE.
Stage 6 — query
Generate a library of validated Cypher queries covering the use-case.
Step Q1 — Fetch current schema
# Preferred: use neo4j-mcp get-schema tool
# Fallback:
source .env
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CYPHER 25 CALL db.labels() YIELD label RETURN label"
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CYPHER 25 CALL db.relationshipTypes() YIELD relationshipType RETURN relationshipType"Step Q2 — Check plugin availability
source .env
# GDS — Aura Free does NOT have GDS
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CALL gds.version() YIELD version RETURN version" 2>/dev/null \
&& echo "GDS available" || echo "GDS NOT available — skip GDS queries"
# Check for vector index
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CYPHER 25 SHOW VECTOR INDEXES YIELD name RETURN name"Only generate GDS or vector queries if plugin/index confirmed present.
Step Q3 — Required query set
The gate requires ≥5 queries, ≥2 of which are traversal queries (contain a relationship pattern in MATCH).
Always include (all experience levels)
Q1 — Overview (metadata, not traversal)
// Q1: Node counts by label
CYPHER 25
MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count ORDER BY count DESC;Q2 — Direct connections (traversal, 1-hop) ← use-case specific
// Q2: [Traversal] Find direct neighbors of a given <Label>
// $id = primary key of the starting node
CYPHER 25
MATCH (n:<Label> {id: $id})-[r]->(m)
RETURN type(r) AS relationship, labels(m)[0] AS type, m.name AS name
LIMIT 25;Q3 — Indirect pattern (traversal, 2+ hops) ← core use-case query
// Q3: [Traversal] <Use-case-specific multi-hop query>
// e.g. for recommendations: friends-of-friends not yet followed
CYPHER 25
MATCH (me:<Label> {id: $id})-[:<REL>]->(intermediate)-[:<REL>]->(target)
WHERE NOT exists { (me)-[:<REL>]->(target) } AND me <> target
WITH target, count(intermediate) AS strength
ORDER BY strength DESC LIMIT 10
RETURN target.name AS recommendation, strength;Q4 — Aggregation (business metric)
// Q4: Top <entities> by <metric>
CYPHER 25
MATCH (n:<Label>)
RETURN n.name AS name, n.<metric> AS value
ORDER BY value DESC LIMIT 20;Q5 — Filtered list
// Q5: <Entities> matching criteria
// $threshold = filter value
CYPHER 25
MATCH (n:<Label>)
WHERE n.<property> > $threshold
RETURN n.name AS name, n.<property> AS value
ORDER BY value DESC LIMIT 50;Intermediate / advanced additions
Q6 — Aggregation within traversal
// Q6: [Traversal] Average <metric> across connected entities
CYPHER 25
MATCH (a:<LabelA>)-[:<REL>]->(b:<LabelB>)
RETURN a.name AS entity, avg(b.<metric>) AS avgMetric, count(b) AS connections
ORDER BY avgMetric DESC LIMIT 20;Q7 — Vector similarity search (only if vector index confirmed)
// Q7: Semantic similarity search
// $embedding = query vector (inject at runtime)
CYPHER 25
MATCH (chunk)
SEARCH chunk IN (
VECTOR INDEX chunk_embeddings
FOR $embedding
LIMIT 5
) SCORE AS score
MATCH (chunk)<-[:HAS_CHUNK]-(doc:Document)
RETURN chunk.text AS text, doc.title AS source, score
ORDER BY score DESC;Q8 — Fulltext search (only if fulltext index confirmed)
// Q8: Fulltext search across <Label>
// $searchTerm = user search input
CYPHER 25
CALL db.index.fulltext.queryNodes('<label>_search', $searchTerm)
YIELD node, score
RETURN node.name AS name, node.<prop> AS description, score
ORDER BY score DESC LIMIT 20;Step Q4 — Validate all queries in one batch call
Do not run cypher-shell once per query — batch into a single Python script using the Query API helper:
# Run from the work directory:
# python3 -c "$(cat ${CLAUDE_SKILL_DIR}/scripts/validate_queries.py)"
# Or copy the script and run it directly.The skill ships a validation helper at ${CLAUDE_SKILL_DIR}/scripts/validate_queries.py. Run it after drafting queries/queries.cypher:
python3 "${CLAUDE_SKILL_DIR}/scripts/validate_queries.py"Reads queries/queries.cypher, substitutes $param placeholders with safe defaults, runs all queries in a single driver session, prints a pass/fail table. Keep queries marked ✓; remove or fix ✗.
A query returning 0 rows is acceptable if the schema confirms the pattern exists.
Step Q5 — Write queries/queries.cypher
mkdir -p queries// ============================================================
// Query Library — <DOMAIN> / <USE_CASE>
// Generated by neo4j-getting-started-skill
// Run with: cypher-shell -a $NEO4J_URI -u $NEO4J_USERNAME -p $NEO4J_PASSWORD
// or paste into Neo4j Browser / Workspace
// ============================================================
// Q1: Overview — node counts
CYPHER 25
MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count ORDER BY count DESC;
// ... additional queriesQuery authoring rules
- Every query starts with
CYPHER 25 - Always specify node labels — no label-free
MATCH (n) - Use
$paramplaceholders for filterable values; provide a concrete default in the comment in this exact format so the validator can auto-substitute:
// $accountId = "p1" ← validator replaces $accountId with "p1"
// $limit = 10 ← validator replaces $limit with 10Use $id or $<label>Id naming (e.g. $accountId, $personId) — the validator auto-substitutes these with the sample_id from progress.md without needing a comment. Never write // $param = description text — the validator captures the first word as the default, which is wrong.
- Every read query has
LIMIT - Labels and property names are case-sensitive — match schema exactly
- No GDS/APOC unless confirmed available
On Completion — write to progress.md
### 6-query
status: done
queries_total=<number>
traversal_queries=<number>
queries_returning_rows=<number>
files=queries/queries.cypherCompletion condition
queries/queries.cypherhas ≥5 queries- ≥2 are traversal queries (contain
->or<-relationship pattern in MATCH) - ≥3 execute and return ≥1 row on the imported data
Stage 7 — build
Generate a runnable application, dashboard, or agent integration.
v1: Python only. JavaScript in phase 2.
Virtual environment (required — always use .venv)
Modern Python (3.12+) forbids global pip install — always use .venv from stage 0.
.venv/bin/pip install -r requirements.txtNever use bare pip install or python3 — always .venv/bin/pip and .venv/bin/python3.
Always include in requirements.txt
neo4j-rust-ext>=0.0.1
python-dotenv>=1.0.0CRITICAL: Do not add neo4j as a standalone dependency — it is a transitive dependency of neo4j-rust-ext. If you write neo4j>=... in requirements.txt, delete it and replace with neo4j-rust-ext>=0.0.1.
Add neo4j-viz>=1.0.0 to requirements.txt for Path A (notebook) and Path B (Streamlit) — it is used for graph visualization and must be listed explicitly.
Path selection
APP_TYPE=notebook → Path A: Jupyter notebook
APP_TYPE=streamlit → Path B: Streamlit dashboard
APP_TYPE=fastapi → Path C: FastAPI backend
APP_TYPE=graphrag → Path D: GraphRAG pipeline
APP_TYPE=explore-only → skip build; output queries.cypher + README only
APP_TYPE=mcp → Path E: neo4j-mcp configurationPath A — Jupyter Notebook
Two-step: test Python snippets first, then compose into notebook. Avoids writing a large notebook only to discover connection or query errors.
Step A0 — Smoke-test key snippets in isolation
Before writing the notebook, verify the critical pieces work:
# Test: connection + use-case query (run with python3, not jupyter)
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os, pandas as pd
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
driver.verify_connectivity()
print("✓ Connected")
# Test the core use-case query (adapt to domain)
records, _, _ = driver.execute_query("""
CYPHER 25
MATCH (me:Person {id: $id})-[:FOLLOWS]->(f)-[:FOLLOWS]->(fof)
WHERE NOT exists { (me)-[:FOLLOWS]->(fof) } AND me <> fof
WITH fof, count(DISTINCT f) AS mutual
ORDER BY mutual DESC LIMIT 10
RETURN fof.name AS recommendation, mutual
""", id="p1", database_="neo4j")
df = pd.DataFrame([r.data() for r in records])
assert len(df) > 0, "No recommendations returned — check data and query"
print(f"✓ Use-case query works: {len(df)} recommendations")
driver.close()Run: .venv/bin/python3 /tmp/smoke_test.py. Only proceed to notebook composition once this passes.
Step A1 — Compose notebook.ipynb
Required cells (one focus per cell):
1. Setup — imports + .env loading via python-dotenv 2. Connection — create driver, verify connectivity 3. Schema — CALL db.labels() etc., display as DataFrame 4. Per-query cells — one cell per query from queries/queries.cypher, display as DataFrame 5. Graph visualization — REQUIRED, do not skip — interactive graph using neo4j-viz 6. Use-case answer cell — use the query verified in Step A0; include assertion + plot
# Cell: Setup (run once if packages are missing)
# %pip install -q neo4j-rust-ext python-dotenv pandas matplotlib neo4j-viz# Cell: Graph Visualization ← REQUIRED — this is the "it clicks" moment for users
from neo4j_viz.neo4j import from_neo4j
from neo4j import RoutingControl
result = driver.execute_query(
"CYPHER 25 MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 50",
routing_=RoutingControl.READ,
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
vg = from_neo4j(result)
vg.color_nodes(field="caption")
vg.render()# Cell: Connection
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os, pandas as pd
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
driver.verify_connectivity()
print("✓ Connected")
def run_query(q, params={}):
records, _, _ = driver.execute_query(q, parameters_=params,
database_=os.environ.get("NEO4J_DATABASE","neo4j"))
return pd.DataFrame([r.data() for r in records])# Cell: Use-case answer (adapt to domain)
df = run_query("""
CYPHER 25
MATCH (me:Person {id: $id})-[:FOLLOWS]->(f)-[:FOLLOWS]->(fof)
WHERE NOT exists { (me)-[:FOLLOWS]->(fof) } AND me <> fof
WITH fof, count(DISTINCT f) AS mutual
ORDER BY mutual DESC LIMIT 10
RETURN fof.name AS recommendation, mutual
""", {"id": "1"})
assert len(df) > 0, "No recommendations — check import and traversal query"
df.plot(kind='barh', x='recommendation', y='mutual', title='Recommendations')Validate: python3 -m json.tool notebook.ipynb > /dev/null && echo "✓ Valid notebook"
Install and run:
.venv/bin/pip install -r requirements.txt
.venv/bin/jupyter notebook notebook.ipynbAdd to requirements.txt:
jupyter>=1.0.0
ipykernel>=6.0.0
pandas>=2.0.0
matplotlib>=3.0.0
neo4j-viz>=1.0.0Path B — Streamlit Dashboard
If DATA_SOURCE=documents: use the GraphRAG chatbot template from ${CLAUDE_SKILL_DIR}/references/capabilities/kg-from-documents.md Step K7 — it uses VectorCypherRetriever to ground answers in ingested document chunks.
Generate app.py (generic dashboard for non-documents data sources):
import streamlit as st
from neo4j import GraphDatabase, RoutingControl
from neo4j_viz.neo4j import from_neo4j
from dotenv import load_dotenv
import os, pandas as pd
load_dotenv()
@st.cache_resource
def get_driver():
return GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
def run_query(q, params={}):
records, _, _ = get_driver().execute_query(
q, parameters_=params,
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
return pd.DataFrame([r.data() for r in records])
st.title(f"{DOMAIN} — {USE_CASE}")
# Sidebar controls
limit = st.sidebar.slider("Results limit", 5, 100, 20)
# Section 1: Overview
st.header("Database Overview")
df = run_query("CYPHER 25 MATCH (n) RETURN labels(n)[0] AS label, count(n) AS count")
st.bar_chart(df.set_index("label"))
# Section 2: Graph visualization — REQUIRED, do not skip
st.header("Graph Visualization")
result = get_driver().execute_query(
"CYPHER 25 MATCH (n)-[r]->(m) RETURN n, r, m LIMIT 50",
routing_=RoutingControl.READ,
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
vg = from_neo4j(result)
vg.color_nodes(field="caption")
# render() returns IPython.display.HTML — extract .data for Streamlit
st.components.v1.html(vg.render().data, height=500, scrolling=True)
# Section 3: Use-case answer (adapt to domain)
st.header("<Use-case headline>")
df2 = run_query("<traversal query from queries.cypher>", {"limit": limit})
st.dataframe(df2)
assert not df2.empty, "Query returned no results"Add to requirements.txt:
streamlit>=1.30.0
pandas>=2.0.0
neo4j-viz>=1.0.0Install and run:
.venv/bin/pip install -r requirements.txt
.venv/bin/streamlit run app.pyValidate: .venv/bin/python3 -m py_compile app.py && echo "✓ Syntax OK"
Path C — FastAPI Backend
Step C0 — Smoke-test connection before writing the app
# Save as /tmp/smoke_test.py and run: .venv/bin/python3 /tmp/smoke_test.py
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
driver.verify_connectivity()
records, _, _ = driver.execute_query(
"MATCH (n) RETURN count(n) AS total",
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
assert records[0]["total"] > 0, "Database is empty — check load stage"
print(f"✓ Connected. {records[0]['total']} nodes in DB")
driver.close()Only proceed once this passes.
Step C1 — Generate main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os
load_dotenv()
_driver = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global _driver
_driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
_driver.verify_connectivity()
yield
_driver.close()
app = FastAPI(title=f"{DOMAIN} API — {USE_CASE}", lifespan=lifespan)
def driver():
return _driver
@app.get("/health")
def health():
records, _, _ = driver().execute_query(
"MATCH (n) RETURN count(n) AS total",
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
return {"status": "ok", "total_nodes": records[0]["total"]}
@app.get("/<entities>")
def list_entities(limit: int = 20):
records, _, _ = driver().execute_query(
"CYPHER 25 MATCH (n:<Label>) RETURN n.id AS id, n.name AS name LIMIT $limit",
limit=limit, database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
return [dict(r) for r in records]
@app.get("/<entities>/{id}/recommendations")
def recommendations(id: str, limit: int = 10):
records, _, _ = driver().execute_query(
"<traversal query from queries.cypher>",
id=id, limit=limit,
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
return [dict(r) for r in records]IMPORTANT — Cypher query parameter rule: Named query parameters ($limit, $id, etc.) are passed as keyword arguments directly to execute_query(). Do NOT use limit_= (that is a driver keyword for the built-in limit_ option, not a Cypher parameter). Use limit=limit, id=id, etc.
IMPORTANT — Avoid cross-product inflation in aggregate queries: When computing counts across multiple optional relationships, use COUNT subqueries instead of sequential OPTIONAL MATCH:
// BAD — inflates counts via cross-product:
MATCH (c:Customer {id: $id})
OPTIONAL MATCH (c)-[:PLACED]->(o:Order)
OPTIONAL MATCH (o)-[:CONTAINS]->(p:Product)
RETURN count(o) AS orders, count(p) AS products
// GOOD — independent counts via subqueries:
MATCH (c:Customer {id: $id})
RETURN COUNT { (c)-[:PLACED]->(:Order) } AS orders,
COUNT { (c)-[:PLACED]->(:Order)-[:CONTAINS]->(:Product) } AS productsStep C2 — Validate and run
Add to requirements.txt:
fastapi>=0.110.0
uvicorn>=0.29.0Install, validate, and run:
.venv/bin/pip install -r requirements.txt
.venv/bin/python3 -m py_compile main.py && echo "✓ Syntax OK"
.venv/bin/uvicorn main:app --reloadDocs: http://localhost:8000/docs
Smoke-test the running app:
curl -s http://localhost:8000/health | python3 -m json.toolAssert total_nodes > 0 in the response.
Path D — GraphRAG Pipeline
If DATA_SOURCE=documents: full pipeline in ${CLAUDE_SKILL_DIR}/references/capabilities/kg-from-documents.md (Steps K7/K8). Use the Streamlit chatbot template (Step K7) or ToolsRetriever (Step K8).
For a standalone smoke-test script graphrag_app.py:
from neo4j_graphrag.retrievers import VectorCypherRetriever
from neo4j_graphrag.generation import GraphRAG
from neo4j_graphrag.embeddings import OpenAIEmbeddings
from neo4j_graphrag.llm import OpenAILLM
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os
load_dotenv()
driver = GraphDatabase.driver(os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"]))
embedder = OpenAIEmbeddings(model=os.environ.get("EMBEDDING_MODEL", "text-embedding-3-small"))
llm = OpenAILLM(model_name=os.environ.get("LLM_MODEL", "gpt-5.4-mini"))
# SimpleKGPipeline stores entities as :__KGBuilder__ nodes connected via FROM_CHUNK.
# Always inspect actual schema after ingestion: db.schema.visualization()
retrieval_query = """
OPTIONAL MATCH (entity:__KGBuilder__)-[:FROM_CHUNK]->(node)
RETURN node.text AS chunk_text,
collect(DISTINCT entity.name)[..5] AS entities,
score
ORDER BY score DESC
"""
retriever = VectorCypherRetriever(
driver=driver,
index_name="chunk_embeddings",
retrieval_query=retrieval_query,
embedder=embedder,
neo4j_database=os.environ.get("NEO4J_DATABASE", "neo4j"),
)
rag = GraphRAG(retriever=retriever, llm=llm)
if __name__ == "__main__":
query = input("Ask a question: ")
response = rag.search(query_text=query, retriever_config={"top_k": 5}, return_context=True)
assert response.answer, "GraphRAG returned empty — check embeddings and vector index are ONLINE"
print(response.answer)Add to requirements.txt:
neo4j-graphrag[openai]>=1.13.0Path E — MCP Integration
Install neo4j-mcp binary (done in prerequisites). Write config files:
For Claude Desktop (~/Library/Application Support/Claude/claude_desktop_config.json on macOS, %APPDATA%\Claude\claude_desktop_config.json on Windows):
{
"mcpServers": {
"neo4j": {
"command": "/absolute/path/to/neo4j-mcp",
"env": {
"NEO4J_URI": "<from .env>",
"NEO4J_USERNAME": "neo4j",
"NEO4J_PASSWORD": "<from .env>",
"NEO4J_DATABASE": "neo4j"
}
}
}
}For Claude Code (.claude/settings.json in project root):
{
"mcpServers": {
"neo4j": {
"command": "./neo4j-mcp",
"env": {
"NEO4J_URI": "<from .env>",
"NEO4J_USERNAME": "neo4j",
"NEO4J_PASSWORD": "<from .env>",
"NEO4J_DATABASE": "neo4j"
}
}
}
}Available MCP tools after restart: read-cypher, write-cypher, get-schema, list-gds-procedures.
Tell user: "Restart Claude Desktop or Claude Code — the neo4j server will appear as available tools."
For read-only mode (recommended for production/shared DBs), add:
"NEO4J_READ_ONLY": "true"On Completion — write to progress.md
### 7-build
status: done
artifact=<filename, e.g. notebook.ipynb or app.py>
app_type=<notebook|streamlit|fastapi|graphrag|mcp>
run_command=<e.g. ".venv/bin/jupyter notebook notebook.ipynb" or ".venv/bin/streamlit run app.py">
files=<artifact filename>,requirements.txtCompletion condition
- At least one artifact exists and passes syntax check (or is valid JSON for notebooks)
- At least one cell / endpoint / function returns non-empty results for the use-case query
requirements.txtwritten- MCP config written to correct location (if
APP_TYPE=mcpor requested) - `README.md` written — required final output; follow the README template in
SKILL.md(Final Summary section). Fill every placeholder fromprogress.mdand the actual generated files. Do not skip.
Error recovery
- App returns empty results → verify
loadstage completed, check query parameter names match schema - Import error → check
requirements.txt, run.venv/bin/pip install -r requirements.txt - MCP not appearing in Claude → verify absolute path to binary; for Claude Desktop restart the app; for Claude Code run
/reloador restart Claude Code
Capability — cypher-authoring
Guidelines for generating correct Cypher 25 queries.
For deep Cypher authoring, load neo4j-cypher-authoring-skill if available.
When to use neo4j-cypher-authoring-skill
If neo4j-cypher-authoring-skill is available, defer all Cypher generation to it — it has comprehensive Cypher 25 rules, QPE handling, and a schema-first protocol that produces higher-quality queries.
To check:
ls ../neo4j-cypher-authoring-skill/SKILL.md 2>/dev/null && echo "Available" || echo "Not found"If available: --append-system-prompt ../neo4j-cypher-authoring-skill/SKILL.md
Minimum rules for Cypher generated in this skill
Mandatory
- Every query starts with
CYPHER 25 - Always specify node labels — never
MATCH (n)without a label except for counts - Always add
LIMITto read queries - Use
$paramplaceholders for user-supplied values - Labels and property names are case-sensitive — match schema exactly
- Use
MERGEnotCREATEfor idempotent writes IS NULL/IS NOT NULL— never= null
MERGE safety
// Nodes: always specify the primary key in the MERGE pattern
MERGE (p:Person {id: $id})
SET p.name = $name, p.updatedAt = datetime();
// Relationships: always MATCH both endpoints first, then MERGE the relationship
MATCH (a:Person {id: $fromId})
MATCH (b:Person {id: $toId})
MERGE (a)-[:FOLLOWS]->(b)Write batching
CYPHER 25
UNWIND $batch AS row
CALL (row) {
MERGE (n:Label {id: row.id})
SET n.name = row.name
} IN TRANSACTIONS OF 500 ROWS;Schema-first protocol
Before writing any MATCH clause: 1. Confirm node labels exist in schema 2. Confirm relationship types exist in schema 3. Confirm property names are spelled correctly 4. Check whether indexes exist for the lookup property
GDS / APOC guard
Only generate GDS or APOC queries after confirming availability:
cypher-shell ... "CALL gds.version() YIELD version" 2>/dev/null || echo "GDS not available"Aura Free has no GDS. Local Docker default image has no GDS unless plugin flag is set.
Vector / fulltext search
Vector (confirmed index exists):
CYPHER 25
MATCH (node)
SEARCH node IN (
VECTOR INDEX index_name
FOR $embedding
LIMIT $topK
) SCORE AS score
RETURN node.text AS text, score ORDER BY score DESC;Fulltext:
CYPHER 25
CALL db.index.fulltext.queryNodes('index_name', $searchTerm)
YIELD node, score
RETURN node.name AS name, score ORDER BY score DESC LIMIT 20;Pattern anti-patterns — always apply these rewrites
Existence checks — use EXISTS subquery, not pattern predicate
// ✗ Wrong — legacy pattern predicate, deprecated in CYPHER 25
WHERE NOT (me)-[:FOLLOWS]->(other)
WHERE (a)-[:KNOWS]->(b)
// ✓ Correct — EXISTS subquery
WHERE NOT exists { (me)-[:FOLLOWS]->(other) }
WHERE exists { (a)-[:KNOWS]->(b) }Inline count — use COUNT subquery, not OPTIONAL MATCH + count(DISTINCT)
// ✗ Wrong — verbose and slower
OPTIONAL MATCH (p)<-[:FOLLOWS]-(follower)
RETURN p.name, count(DISTINCT follower) AS followers
// ✓ Correct — inline count subquery
RETURN p.name, count { (p)<-[:FOLLOWS]-() } AS followersProperty access — defer to the final RETURN, aggregate on nodes
Access properties only after filtering and sorting on the minimal node set. Accessing properties in a WITH that feeds ORDER BY or aggregation forces property reads on more rows than necessary.
// ✗ Wrong — property access before aggregation, sorts/limits on property values
MATCH (me:Person {id: $id})-[:FOLLOWS]->(f)-[:FOLLOWS]->(fof)
WHERE NOT exists { (me)-[:FOLLOWS]->(fof) } AND me <> fof
RETURN fof.name AS recommendation, fof.bio AS bio,
count(DISTINCT f) AS mutualFriends
ORDER BY mutualFriends DESC LIMIT 10
// ✓ Correct — aggregate on nodes, sort/limit, then access properties in final RETURN
MATCH (me:Person {id: $id})-[:FOLLOWS]->(f)-[:FOLLOWS]->(fof)
WHERE NOT exists { (me)-[:FOLLOWS]->(fof) } AND me <> fof
WITH fof, count(DISTINCT f) AS mutualFriends
ORDER BY mutualFriends DESC LIMIT 10
RETURN fof.name AS recommendation, fof.bio AS bio, mutualFriendsApply whenever ORDER BY or LIMIT precedes property access: use WITH node, aggregation ORDER BY ... LIMIT then RETURN node.prop.
Common pitfalls (validated against Neo4j 2026.x / CYPHER 25)
| Wrong | Correct | Note |
|---|---|---|
-- comment | // comment | Cypher uses //, not SQL -- |
OPTIONS { indexConfig: { 'vector.dimensions': 1536 } } | OPTIONS { indexConfig: { \vector.dimensions\: 1536 } } | Map keys in OPTIONS must be backtick identifiers, not strings |
(a)-[:REL*0..5]->(b) | (a) (()-[:REL]->()){0,5} (b) | Use quantified path patterns (QPP) in CYPHER 25, not *min..max |
CALL db.index.vector.queryNodes('idx', k, $vec) YIELD node, score | MATCH (node) SEARCH node IN (VECTOR INDEX idx FOR $vec LIMIT k) SCORE AS score | New SEARCH clause (Neo4j 2026.01+); procedure still works but is deprecated |
driver.execute_query("CALL (row) { ... } IN TRANSACTIONS OF N ROWS") | session.run("CALL (row) { ... } IN TRANSACTIONS OF N ROWS") | CALL {} IN TRANSACTIONS requires an auto-commit (implicit) transaction — execute_query uses a managed transaction and will fail |
CALL { MATCH (n) DETACH DELETE n } IN TRANSACTIONS OF 10000 ROWS | MATCH (n) CALL (n) { DETACH DELETE n } IN TRANSACTIONS OF 1000 ROWS | Pass binding variable (n) into subquery so each node is its own batch row; outer CALL {} with inner MATCH doesn't batch at all — runs one giant tx |
CALL { UNWIND $batch AS row MERGE ... } IN TRANSACTIONS OF 500 ROWS | UNWIND $batch AS row CALL (row) { MERGE ... } IN TRANSACTIONS OF 500 ROWS | Always wrong: IN TRANSACTIONS OF N ROWS batches on rows flowing into the subquery from outside. With UNWIND inside, the whole list runs in one transaction — batching has no effect. Move UNWIND outside and import the variable via CALL (row) { ... } |
WHERE NOT (a)-[:REL]->(b) | WHERE NOT exists { (a)-[:REL]->(b) } | Pattern predicates are deprecated in CYPHER 25 — use EXISTS subquery |
WHERE (a)-[:REL]->(b) | WHERE exists { (a)-[:REL]->(b) } | Same — positive pattern check also needs EXISTS subquery |
OPTIONAL MATCH (n)<-[:REL]-(m) RETURN count(DISTINCT m) | RETURN count { (n)<-[:REL]-() } | Use inline COUNT subquery instead of OPTIONAL MATCH + count(DISTINCT) |
RETURN n.name, count(x) ORDER BY count(x) | WITH n, count(x) AS cnt ORDER BY cnt LIMIT k RETURN n.name, cnt | Access properties after aggregation + sort/limit, not before — avoids reading properties on rows that will be discarded |
Capability — execute-cypher
Three options for running Cypher statements against Neo4j.
Detect and record EXEC_METHOD in the context stage — priority order: 1. mcp — if neo4j-mcp is running as an MCP server in this session 2. cypher-shell — if cypher-shell is on PATH 3. query-api — HTTP fallback, always available when DB is reachable
Store as EXEC_METHOD=mcp|cypher-shell|query-api and use it consistently across all stages.
---
Option 1 — neo4j-mcp (MCP tools)
Use when: neo4j-mcp is configured as an MCP server in this agent session.
Read query:
use tool: read-cypher
params: { query: "CYPHER 25 MATCH (n) RETURN count(n) AS total", params: {} }Write query:
use tool: write-cypher
params: { query: "CYPHER 25 MERGE (n:Label {id: $id}) SET n.name = $name", params: { "id": "1", "name": "Test" } }Schema inspection:
use tool: get-schema---
Option 2 — cypher-shell
Use when: which cypher-shell succeeds.
source .env
# Read query
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
--database "${NEO4J_DATABASE:-neo4j}" \
"CYPHER 25 MATCH (n) RETURN count(n) AS total"
# Run a .cypher file
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
--database "${NEO4J_DATABASE:-neo4j}" \
--file schema.cypher
# Write query
cypher-shell -a "$NEO4J_URI" -u "$NEO4J_USERNAME" -p "$NEO4J_PASSWORD" \
"CYPHER 25 MERGE (n:Label {id: '1'}) SET n.name = 'Test'"---
Option 3 — Neo4j Query API (HTTP)
Use when neither MCP nor cypher-shell is available. Works with curl only — no Neo4j client needed.
source .env
# Derive HTTPS host from bolt URI
HOST=$(echo "$NEO4J_URI" | sed 's|neo4j+s://||;s|bolt://||;s|neo4j://||;s|bolt+s://||')
DB="${NEO4J_DATABASE:-neo4j}"
AUTH="$NEO4J_USERNAME:$NEO4J_PASSWORD"
# Read query
curl -s -X POST "https://${HOST}/db/${DB}/query/v2" \
-H "Content-Type: application/json" \
-u "$AUTH" \
-d '{"statement": "CYPHER 25 MATCH (n) RETURN count(n) AS total"}' \
| python3 -c "import sys,json; d=json.load(sys.stdin); [print(r) for r in d.get('data',{}).get('values',[])]"
# Write query
curl -s -X POST "https://${HOST}/db/${DB}/query/v2" \
-H "Content-Type: application/json" \
-u "$AUTH" \
-d '{"statement": "CYPHER 25 MERGE (n:Label {id: $id}) SET n.name = $name", "parameters": {"id": "1", "name": "Test"}}' \
| python3 -c "import sys,json; print(json.load(sys.stdin))"Note: for local Docker, use http://localhost:7474 instead of https://${HOST}.
---
Python driver (always available when neo4j package installed)
from neo4j import GraphDatabase
from dotenv import load_dotenv
import os
load_dotenv()
driver = GraphDatabase.driver(
os.environ["NEO4J_URI"],
auth=(os.environ["NEO4J_USERNAME"], os.environ["NEO4J_PASSWORD"])
)
# Read
records, summary, keys = driver.execute_query(
"CYPHER 25 MATCH (n:<Label>) RETURN n LIMIT $limit",
limit=20,
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
# Write (use write-transaction for mutations)
driver.execute_query(
"CYPHER 25 MERGE (n:<Label> {id: $id}) SET n.name = $name",
id="1", name="Test",
database_=os.environ.get("NEO4J_DATABASE", "neo4j")
)
driver.close()Capability — mcp-config
Configure the official neo4j-mcp server for different agent environments.
Reference this from the build stage when APP_TYPE=mcp or integration is requested.
Binary location
# Check candidates in priority order
NEO4J_MCP_BIN=$(which neo4j-mcp 2>/dev/null \
|| ls $HOME/bin/neo4j-mcp 2>/dev/null \
|| ls ./neo4j-mcp 2>/dev/null \
|| echo "NOT_FOUND")
echo "neo4j-mcp: $NEO4J_MCP_BIN"Use that absolute path in all config files below.
Claude Desktop
macOS: ~/Library/Application Support/Claude/claude_desktop_config.json Windows: %APPDATA%\Claude\claude_desktop_config.json
{
"mcpServers": {
"neo4j": {
"command": "<absolute-path-to-neo4j-mcp>",
"env": {
"NEO4J_URI": "<NEO4J_URI from .env>",
"NEO4J_USERNAME": "<NEO4J_USERNAME from .env>",
"NEO4J_PASSWORD": "<NEO4J_PASSWORD from .env>",
"NEO4J_DATABASE": "neo4j"
}
}
}
}Claude Code
Project-level (.claude/settings.json — checked in, safe since no secrets if using env vars):
{
"mcpServers": {
"neo4j": {
"command": "./neo4j-mcp",
"env": {
"NEO4J_URI": "<NEO4J_URI>",
"NEO4J_USERNAME": "neo4j",
"NEO4J_PASSWORD": "<NEO4J_PASSWORD>",
"NEO4J_DATABASE": "neo4j"
}
}
}
}User-level (~/.claude/settings.json — applies to all projects):
{
"mcpServers": {
"neo4j-<project-name>": {
"command": "<absolute-path-to-neo4j-mcp>",
"env": {
"NEO4J_URI": "<URI>",
"NEO4J_USERNAME": "neo4j",
"NEO4J_PASSWORD": "<PASSWORD>",
"NEO4J_DATABASE": "neo4j"
}
}
}
}Read-only mode (recommended for production / shared DBs)
Add to env:
"NEO4J_READ_ONLY": "true"This disables the write-cypher tool entirely.
Available MCP tools after restart
| Tool | Description |
|---|---|
get-schema | Introspect node labels, relationship types, property keys |
read-cypher | Execute read-only Cypher |
write-cypher | Execute write Cypher (disabled in read-only mode) |
list-gds-procedures | List GDS procedures (only if GDS installed) |
Verify config is working
After restart, ask: "What node labels are in my Neo4j database?" — should use get-schema automatically.
Related skills
How it compares
Pick neo4j-getting-started-skill over neo4j-cypher-skill when bootstrapping a new graph project end to end, not when tuning queries on an existing database.
FAQ
How many stages does neo4j-getting-started-skill run?
neo4j-getting-started-skill runs eight stages in order: prerequisites, context, provision, model, load, explore, query, and build. Each stage reads its own reference file and can run autonomously or with human-in-the-loop checkpoints.
What time budgets does neo4j-getting-started-skill document?
neo4j-getting-started-skill targets ≤15 minutes for fully autonomous runs and ≤90 minutes when human-in-the-loop checkpoints are used. The pipeline covers Aura provisioning through a runnable app or notebook.
What tasks does neo4j-getting-started-skill not cover?
neo4j-getting-started-skill does not cover standalone Cypher query authoring, driver upgrade migrations, or CLI administration on existing databases. Those tasks route to neo4j-cypher-skill, neo4j-migration-skill, and neo4j-cli skills.
Is Neo4j Getting Started Skill safe to install?
skills.sh reports 2 of 3 security scanners passed. Review the Security Audits panel on this page before installing in production.