mirror of
https://github.com/bleak-ai/gcontext.git
synced 2026-08-11 13:19:23 +02:00
gcontext 0.3.0: agent state in a folder, served at a URL
Local-first state manager for AI agents. One HTTP server (gcontext up), any MCP runtime attaches by URL. Context ledger (nothing reaches the agent invisibly), secrets isolation (names visible, values injected and scrubbed), flows (multi-step work computed purely from files), archive convention. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
bernatsampera
@@ -0,0 +1,9 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.venv/
|
||||
dist/
|
||||
*.egg-info/
|
||||
.pytest_cache/
|
||||
|
||||
# secret values never leave the machine
|
||||
secrets.env
|
||||
@@ -0,0 +1,21 @@
|
||||
MIT License
|
||||
|
||||
Copyright (c) 2026 bleak-ai
|
||||
|
||||
Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
of this software and associated documentation files (the "Software"), to deal
|
||||
in the Software without restriction, including without limitation the rights
|
||||
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
copies of the Software, and to permit persons to whom the Software is
|
||||
furnished to do so, subject to the following conditions:
|
||||
|
||||
The above copyright notice and this permission notice shall be included in all
|
||||
copies or substantial portions of the Software.
|
||||
|
||||
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
|
||||
SOFTWARE.
|
||||
@@ -0,0 +1,110 @@
|
||||
# gcontext
|
||||
|
||||
**Your agent's state lives in a folder. It's served at a URL. Any runtime becomes your agent.**
|
||||
|
||||
Claude Code, Codex, Cursor, Claude Desktop: these are runtimes. They read, reason, and act, but they forget everything between sessions. gcontext is the part that persists: the context your agent has learned, the services it can operate, the secrets it can use, the multi-step work in progress. All of it in a plain directory you can version with git.
|
||||
|
||||
gcontext is not a runtime. It has no chat loop, no LLM client, no orchestration engine. It serves your agent's state over MCP, and any MCP client that attaches becomes your agent.
|
||||
|
||||
```bash
|
||||
uv tool install gcontext-ai # or: uv tool install git+https://github.com/bleak-ai/gcontext
|
||||
|
||||
gcontext init my-agent # scaffold the state folder
|
||||
gcontext up my-agent # serve it at http://127.0.0.1:4242/mcp
|
||||
```
|
||||
|
||||
Connect a harness by pasting the URL, once, from any directory:
|
||||
|
||||
```bash
|
||||
claude mcp add --transport http my-agent http://127.0.0.1:4242/mcp
|
||||
```
|
||||
|
||||
The server prints every harness as it attaches. `Ctrl+C` and every harness cleanly loses access. That's the whole model: a server running, and harnesses that connect to it.
|
||||
|
||||
## What's in the folder
|
||||
|
||||
```
|
||||
my-agent/
|
||||
gcontext.yaml # identity: name, description, optional port
|
||||
instructions.md # standing instructions for whatever runtime attaches
|
||||
secrets.env # secret VALUES, gitignored, never leave your machine
|
||||
|
||||
connections/ # services the agent can operate
|
||||
stripe/
|
||||
connection.yaml # declares secret NAMEs and Python deps
|
||||
index.md # how to use the API, patterns that work
|
||||
|
||||
modules/ # knowledge the agent accumulates
|
||||
flows/ # multi-step work, tracked as files (see below)
|
||||
archive/ # anything moved here is out of context, still readable
|
||||
```
|
||||
|
||||
Markdown is the context. YAML is the config. The folder is the agent.
|
||||
|
||||
## The three ideas
|
||||
|
||||
### 1. Nothing reaches the agent invisibly
|
||||
|
||||
The **context ledger** enumerates every pipe that inserts context into the agent, each marked `loaded`, `on demand`, `skipped`, or `UNCONTROLLED` (runtime-owned). See it anytime:
|
||||
|
||||
```bash
|
||||
gcontext context my-agent
|
||||
```
|
||||
|
||||
If gcontext feeds something to the agent, it's on that list. No hidden injection, ever.
|
||||
|
||||
### 2. Secrets: names visible, values never
|
||||
|
||||
The agent sees secret NAMEs only. Values live in `secrets.env`, get injected as environment variables when a script runs (`run_script` tool, deps preinstalled via uv), and are scrubbed from all output. This never changes.
|
||||
|
||||
### 3. Flows: workflows as files, not engines
|
||||
|
||||
A flow declares which files each step needs and produces:
|
||||
|
||||
```yaml
|
||||
steps:
|
||||
- id: draft
|
||||
needs: [flows/brief/brief.md]
|
||||
produces: [flows/brief/draft.md]
|
||||
instructions: Read the brief, write the draft.
|
||||
```
|
||||
|
||||
Step status is computed purely from the filesystem, make-style: `blocked` (a need is missing), `ready` (needs exist, produces don't), `stale` (a need changed after the produces), `done`. A runtime completes a step by writing the declared files. There is no executor, no checkpointer, no stored run state: change an upstream file and downstream steps light up as stale. Progress is git-diffable because progress is files.
|
||||
|
||||
```bash
|
||||
gcontext flows my-agent # the board, per step
|
||||
```
|
||||
|
||||
Attached runtimes get the same board via the `flows()` tool, with step instructions surfacing only when a step is actionable.
|
||||
|
||||
## Commands
|
||||
|
||||
| Command | What it does |
|
||||
|---|---|
|
||||
| `gcontext init <dir>` | Scaffold a new agent state folder |
|
||||
| `gcontext up [dir]` | Serve the folder over MCP at a local URL |
|
||||
| `gcontext status [dir]` | Server up? Who is connected? State overview |
|
||||
| `gcontext connect [client]` | Attach instructions for claude, desktop, codex, cursor |
|
||||
| `gcontext context [dir]` | The context ledger |
|
||||
| `gcontext flows [dir]` | The flow boards |
|
||||
| `gcontext chat [dir]` | A dedicated, fully controlled claude session |
|
||||
|
||||
The tools an attached runtime gets: `overview`, `read_context`, `write_context`, `run_script`, `list_connections`, `flows`.
|
||||
|
||||
## Housekeeping without magic
|
||||
|
||||
When accumulated state starts polluting context, move folders into `archive/`:
|
||||
|
||||
```
|
||||
mv my-agent/modules/old-onboarding my-agent/archive/modules/
|
||||
```
|
||||
|
||||
Archived items are never scanned into overviews or counts, stay readable by path, and every summary reports that they exist. The folder move is the entire mechanism. gcontext never archives, deletes, or reorganizes anything by itself.
|
||||
|
||||
## Scope
|
||||
|
||||
Local-first, by design. The server binds `127.0.0.1` with no auth: everything on your machine, nothing exposed. A remote/deployed story (same shape, a URL with a token) is planned but deliberately not in this release.
|
||||
|
||||
## License
|
||||
|
||||
MIT
|
||||
@@ -0,0 +1,30 @@
|
||||
[project]
|
||||
name = "gcontext-ai"
|
||||
version = "0.3.0"
|
||||
description = "Your agent's state in a folder, served at a URL. Any MCP runtime attaches and becomes your agent."
|
||||
readme = "README.md"
|
||||
license = { text = "MIT" }
|
||||
authors = [{ name = "bleak-ai" }]
|
||||
requires-python = ">=3.11"
|
||||
keywords = ["mcp", "agents", "ai", "state", "context"]
|
||||
dependencies = [
|
||||
"fastmcp>=2.0.0",
|
||||
"pyyaml>=6.0",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
Homepage = "https://github.com/bleak-ai/gcontext"
|
||||
Repository = "https://github.com/bleak-ai/gcontext"
|
||||
|
||||
[project.scripts]
|
||||
gcontext = "gcontext.cli:main"
|
||||
|
||||
[dependency-groups]
|
||||
dev = ["pytest>=8.0"]
|
||||
|
||||
[build-system]
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[tool.hatch.build.targets.wheel]
|
||||
packages = ["src/gcontext"]
|
||||
@@ -0,0 +1,572 @@
|
||||
"""gcontext CLI. One server you start, harnesses connect to its URL. State is files."""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
import tempfile
|
||||
import time
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
from . import flows as flows_mod
|
||||
from . import server
|
||||
|
||||
BOLD = "\033[1m"
|
||||
DIM = "\033[2m"
|
||||
GREEN = "\033[32m"
|
||||
YELLOW = "\033[33m"
|
||||
RESET = "\033[0m"
|
||||
|
||||
DEFAULT_PORT = 4242
|
||||
|
||||
STATUS_COLOR = {
|
||||
"loaded": GREEN,
|
||||
"on demand": DIM,
|
||||
"skipped": DIM,
|
||||
"uncontrolled": YELLOW,
|
||||
}
|
||||
|
||||
|
||||
def print_ledger(mode: str):
|
||||
for i, pipe in enumerate(server.build_ledger(mode), 1):
|
||||
color = STATUS_COLOR.get(pipe["status"], "")
|
||||
label = f"{pipe['label']} ".ljust(36, ".")
|
||||
status = pipe["status"].upper() if pipe["status"] == "uncontrolled" else pipe["status"]
|
||||
print(f" {i}. [{pipe['id']}] {label} {color}{status}{RESET} {DIM}{pipe['detail']}{RESET}")
|
||||
|
||||
|
||||
INIT_GCONTEXT_YAML = """\
|
||||
name: {name}
|
||||
description: Describe what this agent is for.
|
||||
# port: 4242
|
||||
"""
|
||||
|
||||
INIT_INSTRUCTIONS = """\
|
||||
# Instructions
|
||||
|
||||
You are the agent for this gcontext project. Your state lives in this folder:
|
||||
read it with read_context, keep it current with write_context.
|
||||
|
||||
- Call overview() first to see connections, modules, flows, and the context ledger.
|
||||
- Call flows() to see multi-step work and what is actionable right now.
|
||||
- Use run_script for anything that needs an API: secrets are injected as env
|
||||
vars (you only ever see their names), deps are preinstalled.
|
||||
- Record what you learn: update the relevant index.md or module so the next
|
||||
session starts smarter than this one.
|
||||
"""
|
||||
|
||||
INIT_SECRETS = """\
|
||||
# Secret VALUES live here and never leave this machine (this file is gitignored).
|
||||
# Each connection's connection.yaml declares which NAMEs it needs.
|
||||
# EXAMPLE_API_KEY=...
|
||||
"""
|
||||
|
||||
INIT_AGENT_GITIGNORE = """\
|
||||
secrets.env
|
||||
.venv/
|
||||
"""
|
||||
|
||||
INIT_CONNECTION_YAML = """\
|
||||
name: httpbin
|
||||
description: Example connection with no secrets, for trying run_script.
|
||||
secrets: []
|
||||
deps:
|
||||
- requests
|
||||
"""
|
||||
|
||||
INIT_CONNECTION_INDEX = """\
|
||||
# httpbin
|
||||
|
||||
A dummy connection to try run_script without needing any secret. Replace it
|
||||
with a real service: declare secret NAMEs and deps in connection.yaml, put
|
||||
values in secrets.env, and document usage patterns here.
|
||||
|
||||
```python
|
||||
import requests
|
||||
print(requests.get("https://httpbin.org/get").json()["url"])
|
||||
```
|
||||
"""
|
||||
|
||||
INIT_FLOW_YAML = """\
|
||||
name: demo-brief
|
||||
description: Demo flow. Capture a brief, draft from it, then finalize.
|
||||
steps:
|
||||
- id: capture
|
||||
description: Capture what the user wants covered
|
||||
produces:
|
||||
- flows/demo-brief/brief.md
|
||||
instructions: |
|
||||
Ask the user what they want a short write-up about and save the answers
|
||||
to flows/demo-brief/brief.md as a short markdown brief.
|
||||
|
||||
- id: draft
|
||||
description: Draft the write-up from the brief
|
||||
needs:
|
||||
- flows/demo-brief/brief.md
|
||||
produces:
|
||||
- flows/demo-brief/draft.md
|
||||
instructions: |
|
||||
Read the brief and write a first draft to flows/demo-brief/draft.md.
|
||||
|
||||
- id: finalize
|
||||
description: Polish the draft into the final version
|
||||
needs:
|
||||
- flows/demo-brief/brief.md
|
||||
- flows/demo-brief/draft.md
|
||||
produces:
|
||||
- flows/demo-brief/final.md
|
||||
instructions: |
|
||||
Tighten the draft into flows/demo-brief/final.md. If the brief changed
|
||||
since the draft, this step shows as stale: redo it from the current brief.
|
||||
"""
|
||||
|
||||
|
||||
def cmd_init(args):
|
||||
target = Path(args.directory).resolve()
|
||||
if target.exists() and any(target.iterdir()):
|
||||
print(f"Error: {target} already exists and is not empty.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
name = target.name
|
||||
files = {
|
||||
"gcontext.yaml": INIT_GCONTEXT_YAML.format(name=name),
|
||||
"instructions.md": INIT_INSTRUCTIONS,
|
||||
"secrets.env": INIT_SECRETS,
|
||||
".gitignore": INIT_AGENT_GITIGNORE,
|
||||
"connections/httpbin/connection.yaml": INIT_CONNECTION_YAML,
|
||||
"connections/httpbin/index.md": INIT_CONNECTION_INDEX,
|
||||
"flows/demo-brief/flow.yaml": INIT_FLOW_YAML,
|
||||
"modules/.gitkeep": "",
|
||||
"archive/.gitkeep": "",
|
||||
}
|
||||
for rel, content in files.items():
|
||||
f = target / rel
|
||||
f.parent.mkdir(parents=True, exist_ok=True)
|
||||
f.write_text(content)
|
||||
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} created {name} at {target}")
|
||||
print()
|
||||
print("The folder IS your agent's state: version it with git, edit it freely.")
|
||||
print()
|
||||
print("Next steps:")
|
||||
print(f" 1. gcontext up {args.directory} start the server")
|
||||
print(f" 2. gcontext connect claude attach a harness (or: desktop, codex, cursor)")
|
||||
print(f" 3. gcontext chat {args.directory} or talk to a dedicated, fully controlled session")
|
||||
print()
|
||||
print(f"{DIM}See what reaches the agent, anytime: gcontext context {args.directory}{RESET}")
|
||||
|
||||
|
||||
def find_project_dir(path: str | None) -> Path:
|
||||
p = Path(path).resolve() if path else Path.cwd()
|
||||
if (p / "gcontext.yaml").exists():
|
||||
return p
|
||||
print(f"Error: no gcontext.yaml found in {p}", file=sys.stderr)
|
||||
print("Run from a gcontext project directory or pass the path as an argument.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def resolve_port(args) -> int:
|
||||
if getattr(args, "port", None):
|
||||
return args.port
|
||||
config = server._load_gcontext_yaml()
|
||||
return int(config.get("port", DEFAULT_PORT))
|
||||
|
||||
|
||||
def server_url(port: int) -> str:
|
||||
return f"http://127.0.0.1:{port}/mcp"
|
||||
|
||||
|
||||
def fetch_status(port: int) -> dict | None:
|
||||
"""Query the running server. None means nothing is listening."""
|
||||
try:
|
||||
with urllib.request.urlopen(f"http://127.0.0.1:{port}/status", timeout=2) as resp:
|
||||
return json.loads(resp.read())
|
||||
except (urllib.error.URLError, OSError, ValueError):
|
||||
return None
|
||||
|
||||
|
||||
def cmd_up(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
config = server._load_gcontext_yaml()
|
||||
name = config.get("name", project_dir.name)
|
||||
port = resolve_port(args)
|
||||
url = server_url(port)
|
||||
|
||||
running = fetch_status(port)
|
||||
if running is not None:
|
||||
print(f"Error: something already listens on port {port}", file=sys.stderr)
|
||||
print(f" ({running.get('name', 'unknown')} serving {running.get('project_dir', '?')})", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
server.ensure_venv()
|
||||
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} {name}")
|
||||
print(f"{DIM}State: {project_dir}{RESET}")
|
||||
print()
|
||||
print(f"Serving at {BOLD}{url}{RESET}")
|
||||
print()
|
||||
print("Connect a harness (once per harness, works from any directory):")
|
||||
print(f" Claude Code: claude mcp add --transport http {name} {url}")
|
||||
print(f" Claude Desktop: Settings -> Connectors -> Add custom connector -> {url}")
|
||||
print(f' Cursor: "{name}": {{"url": "{url}"}} in ~/.cursor/mcp.json')
|
||||
print(f' Codex: [mcp_servers.{name}] url = "{url}" in ~/.codex/config.toml')
|
||||
print(" Details: gcontext connect")
|
||||
print()
|
||||
print("Connections appear below as harnesses attach. Ctrl+C stops the server,")
|
||||
print("and every harness cleanly loses access.")
|
||||
print()
|
||||
|
||||
server.mcp.run(
|
||||
transport="http", host="127.0.0.1", port=port, path="/mcp",
|
||||
show_banner=False, log_level="warning",
|
||||
)
|
||||
|
||||
|
||||
def cmd_status(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
|
||||
config = server._load_gcontext_yaml()
|
||||
connections = server._load_connections()
|
||||
secrets = server._load_secrets_env()
|
||||
modules = server._discover_modules()
|
||||
port = resolve_port(args)
|
||||
|
||||
name = config.get("name", project_dir.name)
|
||||
desc = config.get("description", "")
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} {name}")
|
||||
if desc:
|
||||
print(f"{DIM}{desc}{RESET}")
|
||||
print(f"{DIM}State: {project_dir}{RESET}")
|
||||
print()
|
||||
|
||||
live = fetch_status(port)
|
||||
if live is None:
|
||||
print(f"Server: {YELLOW}not running{RESET} {DIM}(start it: gcontext up){RESET}")
|
||||
elif live.get("project_dir") != str(project_dir.resolve()):
|
||||
print(f"Server: {YELLOW}port {port} is serving a different project{RESET}")
|
||||
print(f" {DIM}{live.get('name', '?')} at {live.get('project_dir', '?')}{RESET}")
|
||||
else:
|
||||
print(f"Server: {GREEN}up{RESET} at {server_url(port)}")
|
||||
sessions = live.get("sessions", [])
|
||||
if not sessions:
|
||||
print(f" {DIM}no harness connected yet{RESET}")
|
||||
for s in sessions:
|
||||
print(f" {GREEN}{s['client']}{RESET} {DIM}{s['version']}{RESET} connected {s['connected']} last activity {s['last_seen']}")
|
||||
print()
|
||||
|
||||
instructions = project_dir / "instructions.md"
|
||||
if instructions.exists():
|
||||
lines = len(instructions.read_text().splitlines())
|
||||
print(f"Instructions: instructions.md ({lines} lines)")
|
||||
print()
|
||||
|
||||
print("Connections:")
|
||||
if not connections:
|
||||
print(f" {DIM}none defined{RESET}")
|
||||
for cname, conn in connections.items():
|
||||
missing = [s for s in conn.secrets if s not in secrets or not secrets[s]]
|
||||
if missing:
|
||||
print(f" {cname}: {YELLOW}missing {', '.join(missing)}{RESET}")
|
||||
else:
|
||||
filled = len(conn.secrets)
|
||||
print(f" {cname}: {GREEN}ready{RESET} {DIM}({filled}/{filled} secrets){RESET}")
|
||||
print()
|
||||
|
||||
if modules:
|
||||
print("Modules:")
|
||||
for mname, mod in modules.items():
|
||||
suffix = f" {DIM}- {mod.description}{RESET}" if mod.description else ""
|
||||
print(f" {mname}{suffix}")
|
||||
print()
|
||||
|
||||
all_flows = flows_mod.load_flows(project_dir)
|
||||
if all_flows:
|
||||
print("Flows:")
|
||||
for fname, flow in all_flows.items():
|
||||
board = flows_mod.flow_board(project_dir, flow)
|
||||
done = sum(1 for s in board if s["status"] == "done")
|
||||
ready = [s["id"] for s in flows_mod.actionable(board)]
|
||||
if ready:
|
||||
print(f" {fname}: {done}/{len(board)} done, {GREEN}actionable: {', '.join(ready)}{RESET}")
|
||||
else:
|
||||
print(f" {fname}: {done}/{len(board)} done")
|
||||
print(f" {DIM}details: gcontext flows{RESET}")
|
||||
print()
|
||||
|
||||
archived_line = server._archived_line()
|
||||
if archived_line:
|
||||
print(f"{DIM}{archived_line}{RESET}")
|
||||
print()
|
||||
|
||||
print(f"{DIM}No runtime included. Point any MCP client at the URL above.{RESET}")
|
||||
|
||||
|
||||
def cmd_connect(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
config = server._load_gcontext_yaml()
|
||||
name = config.get("name", project_dir.name)
|
||||
port = resolve_port(args)
|
||||
url = server_url(port)
|
||||
|
||||
live = fetch_status(port)
|
||||
if live is None:
|
||||
print(f"{YELLOW}Server not running.{RESET} Start it first, in this or another terminal:")
|
||||
print()
|
||||
print(f" gcontext up {project_dir}")
|
||||
print()
|
||||
|
||||
client = args.client
|
||||
|
||||
if client == "claude":
|
||||
print(f"{BOLD}Claude Code{RESET}")
|
||||
print()
|
||||
print("Run once, from the directory where you use claude (or add --scope user")
|
||||
print("to make it available everywhere):")
|
||||
print()
|
||||
print(f" claude mcp add --transport http {name} {url}")
|
||||
|
||||
elif client == "desktop":
|
||||
print(f"{BOLD}Claude Desktop{RESET}")
|
||||
print()
|
||||
print("Settings -> Connectors -> Add custom connector, then paste:")
|
||||
print()
|
||||
print(f" {url}")
|
||||
|
||||
elif client == "codex":
|
||||
print(f"{BOLD}Codex{RESET}")
|
||||
print()
|
||||
print("Add to ~/.codex/config.toml:")
|
||||
print()
|
||||
print(f"[mcp_servers.{name}]")
|
||||
print(f'url = "{url}"')
|
||||
|
||||
elif client == "cursor":
|
||||
print(f"{BOLD}Cursor{RESET}")
|
||||
print()
|
||||
print("Add to .cursor/mcp.json (project) or ~/.cursor/mcp.json (global):")
|
||||
print()
|
||||
print(json.dumps({"mcpServers": {name: {"url": url}}}, indent=2))
|
||||
|
||||
else:
|
||||
print(f"{BOLD}Any MCP client{RESET}")
|
||||
print()
|
||||
print("gcontext speaks MCP over streamable HTTP. Point your client at:")
|
||||
print()
|
||||
print(f" {url}")
|
||||
|
||||
print()
|
||||
print("Context this client will receive:")
|
||||
print_ledger("mcp")
|
||||
print()
|
||||
print(f"{DIM}Verify anytime with: gcontext status{RESET}")
|
||||
|
||||
|
||||
def cmd_context(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
config = server._load_gcontext_yaml()
|
||||
name = config.get("name", project_dir.name)
|
||||
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} {name}")
|
||||
print(f"{DIM}Every pipe that inserts context into the agent, per mode.{RESET}")
|
||||
print()
|
||||
print(f"{BOLD}gcontext chat{RESET} {DIM}(dedicated claude, fully controlled){RESET}")
|
||||
print_ledger("chat")
|
||||
print()
|
||||
print(f"{BOLD}MCP attach{RESET} {DIM}(any harness pointed at the URL, shared agent){RESET}")
|
||||
print_ledger("mcp")
|
||||
|
||||
|
||||
FLOW_STATUS_COLOR = {
|
||||
"ready": GREEN,
|
||||
"stale": YELLOW,
|
||||
"blocked": DIM,
|
||||
"done": DIM,
|
||||
}
|
||||
|
||||
|
||||
def cmd_flows(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
config = server._load_gcontext_yaml()
|
||||
name = config.get("name", project_dir.name)
|
||||
|
||||
all_flows = flows_mod.load_flows(project_dir)
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} {name}")
|
||||
print(f"{DIM}Flow state is computed from files, nothing else tracks progress.{RESET}")
|
||||
print()
|
||||
|
||||
if not all_flows:
|
||||
print(f"{DIM}No flows defined in flows/*/flow.yaml{RESET}")
|
||||
return
|
||||
|
||||
if args.flow:
|
||||
if args.flow not in all_flows:
|
||||
print(f"Error: no flow named {args.flow}. Available: {', '.join(all_flows)}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
all_flows = {args.flow: all_flows[args.flow]}
|
||||
|
||||
for flow in all_flows.values():
|
||||
board = flows_mod.flow_board(project_dir, flow)
|
||||
done = sum(1 for s in board if s["status"] == "done")
|
||||
print(f"{BOLD}{flow.name}{RESET} {DIM}({done}/{len(board)} done){RESET}")
|
||||
if flow.description:
|
||||
print(f"{DIM}{flow.description}{RESET}")
|
||||
for step in board:
|
||||
color = FLOW_STATUS_COLOR.get(step["status"], "")
|
||||
status = f"{color}{step['status']:<7}{RESET}"
|
||||
print(f" {status} {step['id']}: {step['description']}")
|
||||
if step["status"] == "blocked":
|
||||
print(f" {DIM}waiting on: {', '.join(step['missing'])}{RESET}")
|
||||
elif step["status"] == "ready":
|
||||
print(f" {DIM}complete by writing: {', '.join(step['missing'])}{RESET}")
|
||||
elif step["status"] == "stale":
|
||||
print(f" {YELLOW}{', '.join(step['stale_needs'])} changed after the produces were written{RESET}")
|
||||
print()
|
||||
|
||||
|
||||
CHAT_TOOLS = ",".join(
|
||||
f"mcp__gcontext__{t}"
|
||||
for t in ["overview", "read_context", "write_context", "run_script", "list_connections", "flows"]
|
||||
)
|
||||
|
||||
|
||||
def wait_for_server(port: int, project_dir: Path, timeout: float = 15.0) -> bool:
|
||||
deadline = time.monotonic() + timeout
|
||||
while time.monotonic() < deadline:
|
||||
live = fetch_status(port)
|
||||
if live is not None and live.get("project_dir") == str(project_dir.resolve()):
|
||||
return True
|
||||
time.sleep(0.3)
|
||||
return False
|
||||
|
||||
|
||||
def cmd_chat(args):
|
||||
project_dir = find_project_dir(args.project)
|
||||
server.PROJECT_DIR = project_dir
|
||||
config = server._load_gcontext_yaml()
|
||||
name = config.get("name", project_dir.name)
|
||||
port = resolve_port(args)
|
||||
url = server_url(port)
|
||||
|
||||
print(f"{BOLD}gcontext{RESET} {DIM}-{RESET} {name}")
|
||||
print()
|
||||
print("Context loaded into this session:")
|
||||
print_ledger("chat")
|
||||
print()
|
||||
|
||||
own_server = None
|
||||
live = fetch_status(port)
|
||||
if live is not None and live.get("project_dir") != str(project_dir.resolve()):
|
||||
print(f"Error: port {port} is serving a different project ({live.get('name', '?')})", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
if live is None:
|
||||
print(f"{DIM}Starting server at {url}...{RESET}")
|
||||
own_server = subprocess.Popen(
|
||||
[sys.executable, "-m", "gcontext.cli", "up", str(project_dir), "--port", str(port)],
|
||||
stdout=subprocess.DEVNULL,
|
||||
stderr=subprocess.DEVNULL,
|
||||
)
|
||||
if not wait_for_server(port, project_dir):
|
||||
own_server.terminate()
|
||||
print("Error: server did not come up.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
else:
|
||||
print(f"{DIM}Using the already running server at {url}{RESET}")
|
||||
|
||||
mcp_config = {"mcpServers": {"gcontext": {"type": "http", "url": url}}}
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".json", delete=False, prefix="gcontext-mcp-"
|
||||
) as f:
|
||||
json.dump(mcp_config, f)
|
||||
mcp_config_path = f.name
|
||||
|
||||
cmd = [
|
||||
"claude",
|
||||
"--mcp-config", mcp_config_path,
|
||||
"--strict-mcp-config",
|
||||
"--setting-sources", "",
|
||||
"--allowedTools", CHAT_TOOLS,
|
||||
]
|
||||
instructions = project_dir / "instructions.md"
|
||||
if instructions.exists():
|
||||
cmd.extend(["--system-prompt", instructions.read_text()])
|
||||
|
||||
print(f"{DIM}Starting claude...{RESET}")
|
||||
try:
|
||||
subprocess.run(cmd, cwd=project_dir)
|
||||
finally:
|
||||
Path(mcp_config_path).unlink(missing_ok=True)
|
||||
if own_server is not None:
|
||||
own_server.terminate()
|
||||
try:
|
||||
own_server.wait(timeout=5)
|
||||
except subprocess.TimeoutExpired:
|
||||
own_server.kill()
|
||||
print(f"{DIM}Stopped the session's server.{RESET}")
|
||||
|
||||
|
||||
def main():
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="gcontext",
|
||||
description="Agent state in a folder, served at a URL. Bring your own runtime.",
|
||||
)
|
||||
subparsers = parser.add_subparsers(dest="command")
|
||||
|
||||
init_parser = subparsers.add_parser("init", help="Scaffold a new agent state folder")
|
||||
init_parser.add_argument("directory", help="Directory to create (its name becomes the agent name)")
|
||||
|
||||
def add_common(p):
|
||||
p.add_argument("project", nargs="?", help="Path to gcontext project directory")
|
||||
p.add_argument("--port", type=int, help=f"Server port (default: {DEFAULT_PORT}, or port: in gcontext.yaml)")
|
||||
|
||||
up_parser = subparsers.add_parser("up", help="Start the server. Harnesses connect to its URL")
|
||||
add_common(up_parser)
|
||||
|
||||
status_parser = subparsers.add_parser("status", help="Server up? Who is connected? Plus connections, secrets, modules")
|
||||
add_common(status_parser)
|
||||
|
||||
connect_parser = subparsers.add_parser("connect", help="Show how to point a harness at the server URL")
|
||||
connect_parser.add_argument(
|
||||
"client",
|
||||
nargs="?",
|
||||
default="generic",
|
||||
choices=["claude", "desktop", "codex", "cursor", "generic"],
|
||||
help="Which MCP client to show instructions for",
|
||||
)
|
||||
add_common(connect_parser)
|
||||
|
||||
context_parser = subparsers.add_parser("context", help="Show the context ledger: every pipe into the agent, per mode")
|
||||
add_common(context_parser)
|
||||
|
||||
flows_parser = subparsers.add_parser("flows", help="Show flow boards: step status computed from the files")
|
||||
flows_parser.add_argument("--flow", help="Show a single flow by name")
|
||||
add_common(flows_parser)
|
||||
|
||||
chat_parser = subparsers.add_parser("chat", help="Launch a dedicated claude session against this project")
|
||||
add_common(chat_parser)
|
||||
|
||||
args = parser.parse_args()
|
||||
|
||||
commands = {
|
||||
"init": cmd_init,
|
||||
"up": cmd_up,
|
||||
"status": cmd_status,
|
||||
"connect": cmd_connect,
|
||||
"context": cmd_context,
|
||||
"flows": cmd_flows,
|
||||
"chat": cmd_chat,
|
||||
}
|
||||
if args.command in commands:
|
||||
commands[args.command](args)
|
||||
else:
|
||||
parser.print_help()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
@@ -0,0 +1,113 @@
|
||||
"""Flows: declarative information dependencies, computed from the filesystem.
|
||||
|
||||
A flow is data, not a program. Each step declares which files it needs and
|
||||
which files it produces. Status is a pure function of the filesystem:
|
||||
|
||||
blocked some needed file does not exist yet
|
||||
ready all needs exist, some produced file is missing
|
||||
stale everything exists, but a need is newer than a produce (make semantics)
|
||||
done all produces exist and are up to date
|
||||
|
||||
gcontext never executes a step. A runtime completes a step by writing the
|
||||
declared produces (write_context, or any editor); status recomputes from the
|
||||
files on the next read. There is no run state stored anywhere else.
|
||||
"""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import yaml
|
||||
|
||||
from .models import FlowManifest, FlowStep
|
||||
|
||||
|
||||
def load_flows(project_dir: Path) -> dict[str, FlowManifest]:
|
||||
"""Scan flows/ for subdirectories containing flow.yaml."""
|
||||
flows_dir = project_dir / "flows"
|
||||
if not flows_dir.is_dir():
|
||||
return {}
|
||||
result = {}
|
||||
for item in sorted(flows_dir.iterdir()):
|
||||
flow_file = item / "flow.yaml"
|
||||
if not item.is_dir() or not flow_file.exists():
|
||||
continue
|
||||
data = yaml.safe_load(flow_file.read_text()) or {}
|
||||
manifest = FlowManifest(**data)
|
||||
result[manifest.name] = manifest
|
||||
return result
|
||||
|
||||
|
||||
def step_state(project_dir: Path, step: FlowStep) -> dict:
|
||||
"""Compute a step's status purely from the files it declares."""
|
||||
needs = [(p, project_dir / p) for p in step.needs]
|
||||
produces = [(p, project_dir / p) for p in step.produces]
|
||||
|
||||
missing_needs = [p for p, f in needs if not f.is_file()]
|
||||
if missing_needs:
|
||||
return {"status": "blocked", "missing": missing_needs}
|
||||
|
||||
missing_produces = [p for p, f in produces if not f.is_file()]
|
||||
if missing_produces:
|
||||
return {"status": "ready", "missing": missing_produces}
|
||||
|
||||
if needs and produces:
|
||||
oldest_produce = min(f.stat().st_mtime for _, f in produces)
|
||||
stale_needs = [p for p, f in needs if f.stat().st_mtime > oldest_produce]
|
||||
if stale_needs:
|
||||
return {"status": "stale", "stale_needs": stale_needs}
|
||||
|
||||
return {"status": "done"}
|
||||
|
||||
|
||||
def flow_board(project_dir: Path, flow: FlowManifest) -> list[dict]:
|
||||
"""Every step of a flow with its computed state."""
|
||||
board = []
|
||||
for step in flow.steps:
|
||||
state = step_state(project_dir, step)
|
||||
board.append({
|
||||
"id": step.id,
|
||||
"description": step.description,
|
||||
"needs": step.needs,
|
||||
"produces": step.produces,
|
||||
"instructions": step.instructions,
|
||||
**state,
|
||||
})
|
||||
return board
|
||||
|
||||
|
||||
def actionable(board: list[dict]) -> list[dict]:
|
||||
return [s for s in board if s["status"] in ("ready", "stale")]
|
||||
|
||||
|
||||
def render_flow(project_dir: Path, flow: FlowManifest, with_instructions: bool = True) -> list[str]:
|
||||
"""Plain-text board for one flow. Instructions surface only for actionable steps."""
|
||||
board = flow_board(project_dir, flow)
|
||||
done = sum(1 for s in board if s["status"] == "done")
|
||||
|
||||
lines = [f"## {flow.name} ({done}/{len(board)} done)"]
|
||||
if flow.description:
|
||||
lines.append(flow.description)
|
||||
lines.append("")
|
||||
|
||||
for step in board:
|
||||
lines.append(f"- [{step['status']}] {step['id']}: {step['description']}")
|
||||
if step["needs"]:
|
||||
lines.append(f" needs: {', '.join(step['needs'])}")
|
||||
if step["produces"]:
|
||||
lines.append(f" produces: {', '.join(step['produces'])}")
|
||||
if step["status"] == "blocked":
|
||||
lines.append(f" waiting on: {', '.join(step['missing'])}")
|
||||
if step["status"] == "stale":
|
||||
lines.append(f" stale: {', '.join(step['stale_needs'])} changed after the produces were written")
|
||||
|
||||
ready = actionable(board)
|
||||
if ready and with_instructions:
|
||||
lines.append("")
|
||||
lines.append("Actionable now:")
|
||||
for step in ready:
|
||||
lines.append(f"### {step['id']}")
|
||||
if step["instructions"]:
|
||||
lines.append(step["instructions"].rstrip())
|
||||
missing = step.get("missing") or step["produces"]
|
||||
lines.append(f"Complete it by writing: {', '.join(missing)}")
|
||||
|
||||
return lines
|
||||
@@ -0,0 +1,32 @@
|
||||
"""Schema definitions for gcontext manifests."""
|
||||
|
||||
from pydantic import BaseModel
|
||||
|
||||
|
||||
class ModuleManifest(BaseModel):
|
||||
name: str
|
||||
description: str
|
||||
version: str = "0.1.0"
|
||||
author: str = ""
|
||||
tags: list[str] = []
|
||||
|
||||
|
||||
class ConnectionManifest(BaseModel):
|
||||
name: str
|
||||
description: str = ""
|
||||
secrets: list[str] = []
|
||||
deps: list[str] = []
|
||||
|
||||
|
||||
class FlowStep(BaseModel):
|
||||
id: str
|
||||
description: str = ""
|
||||
needs: list[str] = []
|
||||
produces: list[str] = []
|
||||
instructions: str = ""
|
||||
|
||||
|
||||
class FlowManifest(BaseModel):
|
||||
name: str
|
||||
description: str = ""
|
||||
steps: list[FlowStep] = []
|
||||
@@ -0,0 +1,525 @@
|
||||
"""gcontext MCP server. Reads a project directory and exposes it to any MCP client."""
|
||||
|
||||
import os
|
||||
import sys
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
import tempfile
|
||||
|
||||
import yaml
|
||||
from fastmcp import FastMCP
|
||||
from fastmcp.server.middleware import Middleware
|
||||
from starlette.requests import Request
|
||||
from starlette.responses import JSONResponse
|
||||
|
||||
from . import flows as flows_mod
|
||||
from .models import ConnectionManifest, ModuleManifest
|
||||
|
||||
mcp = FastMCP("gcontext")
|
||||
|
||||
# Set by cli.py before the server starts
|
||||
PROJECT_DIR: Path = Path(".")
|
||||
|
||||
# Live MCP sessions, keyed by session id: {"client": ..., "connected": ..., "last_seen": ...}
|
||||
SESSIONS: dict[str, dict] = {}
|
||||
|
||||
|
||||
def _session_id(context) -> str:
|
||||
ctx = getattr(context, "fastmcp_context", None)
|
||||
return getattr(ctx, "session_id", None) or "session"
|
||||
|
||||
|
||||
class ConnectionTracker(Middleware):
|
||||
"""Records who is connected, straight from the MCP initialize handshake."""
|
||||
|
||||
async def on_initialize(self, context, call_next):
|
||||
params = getattr(context.message, "params", None) or context.message
|
||||
info = getattr(params, "clientInfo", None)
|
||||
client = getattr(info, "name", None) or "unknown client"
|
||||
version = getattr(info, "version", "") or ""
|
||||
now = datetime.now().isoformat(timespec="seconds")
|
||||
SESSIONS[_session_id(context)] = {
|
||||
"client": client,
|
||||
"version": version,
|
||||
"connected": now,
|
||||
"last_seen": now,
|
||||
}
|
||||
print(f" + {client} {version} connected ({now})", file=sys.stderr)
|
||||
return await call_next(context)
|
||||
|
||||
async def on_message(self, context, call_next):
|
||||
session = SESSIONS.get(_session_id(context))
|
||||
if session:
|
||||
session["last_seen"] = datetime.now().isoformat(timespec="seconds")
|
||||
return await call_next(context)
|
||||
|
||||
|
||||
mcp.add_middleware(ConnectionTracker())
|
||||
|
||||
|
||||
@mcp.custom_route("/status", methods=["GET"])
|
||||
async def status_route(request: Request) -> JSONResponse:
|
||||
config = _load_gcontext_yaml()
|
||||
flow_summary = {}
|
||||
for fname, flow in flows_mod.load_flows(PROJECT_DIR).items():
|
||||
board = flows_mod.flow_board(PROJECT_DIR, flow)
|
||||
flow_summary[fname] = {
|
||||
"done": sum(1 for s in board if s["status"] == "done"),
|
||||
"total": len(board),
|
||||
"actionable": [s["id"] for s in flows_mod.actionable(board)],
|
||||
}
|
||||
return JSONResponse({
|
||||
"name": config.get("name", PROJECT_DIR.name),
|
||||
"project_dir": str(PROJECT_DIR.resolve()),
|
||||
"sessions": list(SESSIONS.values()),
|
||||
"flows": flow_summary,
|
||||
})
|
||||
|
||||
|
||||
def _load_gcontext_yaml() -> dict:
|
||||
p = PROJECT_DIR / "gcontext.yaml"
|
||||
if p.exists():
|
||||
return yaml.safe_load(p.read_text()) or {}
|
||||
return {}
|
||||
|
||||
|
||||
def _load_connections() -> dict[str, ConnectionManifest]:
|
||||
"""Scan connections/ for subdirectories containing connection.yaml."""
|
||||
conns_dir = PROJECT_DIR / "connections"
|
||||
if not conns_dir.is_dir():
|
||||
return {}
|
||||
result = {}
|
||||
for item in sorted(conns_dir.iterdir()):
|
||||
if not item.is_dir():
|
||||
continue
|
||||
conn_file = item / "connection.yaml"
|
||||
if not conn_file.exists():
|
||||
continue
|
||||
data = yaml.safe_load(conn_file.read_text()) or {}
|
||||
manifest = ConnectionManifest(**data)
|
||||
result[manifest.name] = manifest
|
||||
return result
|
||||
|
||||
|
||||
def _connection_files(name: str) -> list[str]:
|
||||
"""List non-yaml files in a connection folder."""
|
||||
conn_dir = PROJECT_DIR / "connections" / name
|
||||
if not conn_dir.is_dir():
|
||||
return []
|
||||
files = []
|
||||
for f in sorted(conn_dir.rglob("*")):
|
||||
if f.is_file() and f.name != "connection.yaml":
|
||||
files.append(str(f.relative_to(PROJECT_DIR)))
|
||||
return files
|
||||
|
||||
|
||||
def _load_secrets_env() -> dict[str, str]:
|
||||
env_file = PROJECT_DIR / "secrets.env"
|
||||
if not env_file.exists():
|
||||
return {}
|
||||
pairs = {}
|
||||
for line in env_file.read_text().splitlines():
|
||||
line = line.strip()
|
||||
if not line or line.startswith("#"):
|
||||
continue
|
||||
if "=" in line:
|
||||
key, _, value = line.partition("=")
|
||||
pairs[key.strip()] = value.strip()
|
||||
return pairs
|
||||
|
||||
|
||||
def _archived() -> dict[str, list[str]]:
|
||||
"""Names of archived items per category, from archive/{connections,modules,flows}/.
|
||||
|
||||
Anything under archive/ is never scanned into overview, the ledger counts,
|
||||
or the flow boards. It stays readable by path via read_context. Archiving
|
||||
is a plain folder move; there is no metadata and no automatic behavior.
|
||||
"""
|
||||
result = {}
|
||||
for category in ("connections", "modules", "flows"):
|
||||
d = PROJECT_DIR / "archive" / category
|
||||
if d.is_dir():
|
||||
items = [i.name for i in sorted(d.iterdir()) if i.is_dir()]
|
||||
if items:
|
||||
result[category] = items
|
||||
return result
|
||||
|
||||
|
||||
def _archived_line() -> str:
|
||||
archived = _archived()
|
||||
if not archived:
|
||||
return ""
|
||||
parts = [f"{len(items)} {cat}" for cat, items in archived.items()]
|
||||
return f"archive/: {', '.join(parts)} (not scanned, readable by path)"
|
||||
|
||||
|
||||
def _discover_modules() -> dict[str, ModuleManifest]:
|
||||
"""Scan modules/ for folders with module.yaml."""
|
||||
modules_dir = PROJECT_DIR / "modules"
|
||||
if not modules_dir.is_dir():
|
||||
return {}
|
||||
result = {}
|
||||
for item in sorted(modules_dir.iterdir()):
|
||||
if not item.is_dir():
|
||||
continue
|
||||
manifest_file = item / "module.yaml"
|
||||
if manifest_file.exists():
|
||||
data = yaml.safe_load(manifest_file.read_text()) or {}
|
||||
manifest = ModuleManifest(**data)
|
||||
else:
|
||||
manifest = ModuleManifest(name=item.name, description="")
|
||||
result[manifest.name] = manifest
|
||||
return result
|
||||
|
||||
|
||||
def _module_files(name: str) -> list[str]:
|
||||
"""List content files in a module folder."""
|
||||
mod_dir = PROJECT_DIR / "modules" / name
|
||||
if not mod_dir.is_dir():
|
||||
return []
|
||||
files = []
|
||||
for f in sorted(mod_dir.rglob("*")):
|
||||
if f.is_file() and f.name not in ("module.yaml", ".gitkeep"):
|
||||
files.append(str(f.relative_to(PROJECT_DIR)))
|
||||
return files
|
||||
|
||||
|
||||
SCRIPT_TIMEOUT = 60
|
||||
|
||||
|
||||
def _scrub_output(text: str, secrets: dict[str, str]) -> str:
|
||||
for value in secrets.values():
|
||||
if value and len(value) > 3:
|
||||
text = text.replace(value, "***")
|
||||
return text
|
||||
|
||||
|
||||
def _venv_dir() -> Path:
|
||||
return PROJECT_DIR.resolve() / ".venv"
|
||||
|
||||
|
||||
def _venv_python() -> Path:
|
||||
venv = _venv_dir()
|
||||
if sys.platform == "win32":
|
||||
return venv / "Scripts" / "python.exe"
|
||||
return venv / "bin" / "python"
|
||||
|
||||
|
||||
def _collect_deps() -> set[str]:
|
||||
connections = _load_connections()
|
||||
all_deps = set()
|
||||
for conn in connections.values():
|
||||
for dep in conn.deps:
|
||||
all_deps.add(dep)
|
||||
return all_deps
|
||||
|
||||
|
||||
def ensure_venv() -> None:
|
||||
"""Create the project venv if missing and sync connection deps into it."""
|
||||
venv_dir = _venv_dir()
|
||||
|
||||
if not venv_dir.is_dir():
|
||||
subprocess.run(
|
||||
["uv", "venv", str(venv_dir), "--quiet"],
|
||||
check=True,
|
||||
cwd=str(PROJECT_DIR),
|
||||
)
|
||||
|
||||
all_deps = _collect_deps()
|
||||
if all_deps:
|
||||
subprocess.run(
|
||||
["uv", "pip", "install", "--quiet", "--python", str(_venv_python())]
|
||||
+ sorted(all_deps),
|
||||
check=True,
|
||||
cwd=str(PROJECT_DIR),
|
||||
)
|
||||
|
||||
|
||||
def build_ledger(mode: str) -> list[dict]:
|
||||
"""Every pipe that inserts context into the agent for a mode ('chat' or 'mcp').
|
||||
|
||||
Statuses: loaded (pushed at start), on demand (agent pulls, visible as a
|
||||
tool call), skipped (closed by a launch flag), uncontrolled (runtime-owned).
|
||||
"""
|
||||
instructions = PROJECT_DIR / "instructions.md"
|
||||
connections = _load_connections()
|
||||
modules = _discover_modules()
|
||||
n_files = sum(len(_connection_files(c)) for c in connections)
|
||||
n_files += sum(len(_module_files(m)) for m in modules)
|
||||
|
||||
ledger = []
|
||||
|
||||
if mode == "chat":
|
||||
if instructions.exists():
|
||||
n = len(instructions.read_text().splitlines())
|
||||
ledger.append({"id": "G0", "label": "instructions.md", "detail": f"system prompt ({n} lines)", "status": "loaded"})
|
||||
else:
|
||||
ledger.append({"id": "G0", "label": "instructions.md", "detail": "file missing, no system prompt", "status": "skipped"})
|
||||
else:
|
||||
ledger.append({"id": "G0", "label": "instructions.md", "detail": "not auto-loaded in MCP mode, read it via read_context", "status": "on demand"})
|
||||
|
||||
ledger.append({"id": "G1", "label": "tool descriptions", "detail": "6 gcontext tools, pushed at connect", "status": "loaded"})
|
||||
ledger.append({"id": "G2", "label": "overview()", "detail": "project map, secret status", "status": "on demand"})
|
||||
g3_detail = f"{n_files} files in connections/ + modules/"
|
||||
if _archived():
|
||||
g3_detail += "; archive/ not scanned, readable by path"
|
||||
ledger.append({"id": "G3", "label": "read_context()", "detail": g3_detail, "status": "on demand"})
|
||||
ledger.append({"id": "G4", "label": "list_connections()", "detail": f"{len(connections)} connection(s)", "status": "on demand"})
|
||||
ledger.append({"id": "G5", "label": "run_script() output", "detail": "secret values scrubbed", "status": "on demand"})
|
||||
all_flows = flows_mod.load_flows(PROJECT_DIR)
|
||||
ledger.append({"id": "G6", "label": "flows()", "detail": f"{len(all_flows)} flow(s); step instructions surface only when the step is actionable", "status": "on demand"})
|
||||
|
||||
if mode == "chat":
|
||||
ledger.append({"id": "R1", "label": "claude default system prompt", "detail": "replaced by --system-prompt", "status": "skipped"})
|
||||
ledger.append({"id": "R2", "label": "~/.claude/CLAUDE.md + settings", "detail": "closed via --setting-sources ''", "status": "skipped"})
|
||||
ledger.append({"id": "R3", "label": "other MCP servers", "detail": "closed via --strict-mcp-config", "status": "skipped"})
|
||||
ledger.append({"id": "R4", "label": "claude tool harness", "detail": "runtime-owned", "status": "uncontrolled"})
|
||||
else:
|
||||
ledger.append({"id": "R1", "label": "runtime system prompt", "detail": "runtime-owned", "status": "uncontrolled"})
|
||||
ledger.append({"id": "R2", "label": "user/project CLAUDE.md", "detail": "runtime-owned", "status": "uncontrolled"})
|
||||
ledger.append({"id": "R3", "label": "other MCP servers, skills, memory", "detail": "runtime-owned", "status": "uncontrolled"})
|
||||
|
||||
return ledger
|
||||
|
||||
|
||||
def render_ledger_plain(mode: str) -> list[str]:
|
||||
lines = []
|
||||
for i, pipe in enumerate(build_ledger(mode), 1):
|
||||
label = f"{pipe['label']} ".ljust(36, ".")
|
||||
lines.append(f"{i}. [{pipe['id']}] {label} {pipe['status']}: {pipe['detail']}")
|
||||
return lines
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def overview() -> str:
|
||||
"""Show project info, all connections with their secret status, and all modules with descriptions."""
|
||||
config = _load_gcontext_yaml()
|
||||
connections = _load_connections()
|
||||
secrets = _load_secrets_env()
|
||||
modules = _discover_modules()
|
||||
|
||||
lines = []
|
||||
name = config.get("name", PROJECT_DIR.name)
|
||||
desc = config.get("description", "")
|
||||
lines.append(f"# {name}")
|
||||
if desc:
|
||||
lines.append(desc)
|
||||
lines.append("")
|
||||
|
||||
lines.append("## Context ledger")
|
||||
lines.append("Everything that enters your context from this server, and how:")
|
||||
lines.extend(render_ledger_plain("mcp"))
|
||||
lines.append("")
|
||||
|
||||
instructions = PROJECT_DIR / "instructions.md"
|
||||
if instructions.exists():
|
||||
lines.append(f"System prompt: instructions.md ({len(instructions.read_text().splitlines())} lines)")
|
||||
lines.append("")
|
||||
|
||||
lines.append("## Connections")
|
||||
if not connections:
|
||||
lines.append("No connections defined.")
|
||||
for cname, conn in connections.items():
|
||||
filled = sum(1 for s in conn.secrets if s in secrets and secrets[s])
|
||||
total = len(conn.secrets)
|
||||
status = "ready" if filled == total else f"missing {total - filled} secret(s)"
|
||||
missing = [s for s in conn.secrets if s not in secrets or not secrets[s]]
|
||||
lines.append(f"- **{cname}**: {status}")
|
||||
if conn.description:
|
||||
lines.append(f" {conn.description}")
|
||||
if missing:
|
||||
lines.append(f" Missing: {', '.join(missing)}")
|
||||
if conn.deps:
|
||||
lines.append(f" Deps: {', '.join(conn.deps)}")
|
||||
for f in _connection_files(cname):
|
||||
lines.append(f" - {f}")
|
||||
lines.append("")
|
||||
|
||||
if modules:
|
||||
lines.append("## Modules")
|
||||
for mname, mod in modules.items():
|
||||
tag_str = f" [{', '.join(mod.tags)}]" if mod.tags else ""
|
||||
lines.append(f"- **{mname}** (v{mod.version}){tag_str}")
|
||||
if mod.description:
|
||||
lines.append(f" {mod.description}")
|
||||
for f in _module_files(mname):
|
||||
lines.append(f" - {f}")
|
||||
lines.append("")
|
||||
|
||||
all_flows = flows_mod.load_flows(PROJECT_DIR)
|
||||
if all_flows:
|
||||
lines.append("## Flows")
|
||||
for fname, flow in all_flows.items():
|
||||
board = flows_mod.flow_board(PROJECT_DIR, flow)
|
||||
done = sum(1 for s in board if s["status"] == "done")
|
||||
ready = [s["id"] for s in flows_mod.actionable(board)]
|
||||
ready_str = f", actionable: {', '.join(ready)}" if ready else ""
|
||||
lines.append(f"- **{fname}**: {done}/{len(board)} done{ready_str}")
|
||||
if flow.description:
|
||||
lines.append(f" {flow.description}")
|
||||
lines.append("Call flows() for step details and instructions.")
|
||||
lines.append("")
|
||||
|
||||
archived = _archived()
|
||||
if archived:
|
||||
lines.append("## Archive")
|
||||
for cat, items in archived.items():
|
||||
lines.append(f"- archive/{cat}/: {', '.join(items)}")
|
||||
lines.append("Archived items are never scanned or listed above; read them by path if needed.")
|
||||
|
||||
return "\n".join(lines).rstrip()
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def read_context(path: str) -> str:
|
||||
"""Read a file from the project. Use overview() first to see available files."""
|
||||
target = (PROJECT_DIR / path).resolve()
|
||||
if not target.is_relative_to(PROJECT_DIR.resolve()):
|
||||
return f"Error: path {path} is outside the project directory."
|
||||
if not target.exists():
|
||||
return f"Error: {path} does not exist."
|
||||
if not target.is_file():
|
||||
return f"Error: {path} is not a file."
|
||||
return target.read_text()
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def write_context(path: str, content: str) -> str:
|
||||
"""Write or update a file in the project. Creates parent directories if needed.
|
||||
|
||||
Use this to update connection context docs, create playbooks, write logs, etc.
|
||||
Cannot write to secrets.env or connection.yaml files.
|
||||
|
||||
Args:
|
||||
path: Relative path within the project (e.g. 'modules/support-workflow/playbooks/refund.md')
|
||||
content: The full file content to write.
|
||||
"""
|
||||
target = (PROJECT_DIR / path).resolve()
|
||||
if not target.is_relative_to(PROJECT_DIR.resolve()):
|
||||
return f"Error: path {path} is outside the project directory."
|
||||
if target.name == "secrets.env":
|
||||
return "Error: cannot write to secrets.env through the agent."
|
||||
if target.name == "connection.yaml":
|
||||
return "Error: cannot write to connection.yaml through the agent."
|
||||
|
||||
target.parent.mkdir(parents=True, exist_ok=True)
|
||||
target.write_text(content)
|
||||
return f"Written: {path} ({len(content)} bytes)"
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def run_script(code: str) -> str:
|
||||
"""Run a Python script in the project's .venv with secrets as env vars.
|
||||
|
||||
The .venv has all connection deps pre-installed.
|
||||
Access secrets with os.environ["SECRET_NAME"].
|
||||
Secret values are scrubbed from stdout/stderr before returning.
|
||||
|
||||
Args:
|
||||
code: Python source code to execute.
|
||||
"""
|
||||
secrets = _load_secrets_env()
|
||||
python = _venv_python()
|
||||
|
||||
if not python.exists():
|
||||
ensure_venv()
|
||||
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w", suffix=".py", delete=False, dir=PROJECT_DIR
|
||||
) as f:
|
||||
f.write(code)
|
||||
script_path = f.name
|
||||
|
||||
try:
|
||||
env = os.environ.copy()
|
||||
env.update(secrets)
|
||||
|
||||
result = subprocess.run(
|
||||
[str(python), script_path],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=SCRIPT_TIMEOUT,
|
||||
env=env,
|
||||
cwd=str(PROJECT_DIR),
|
||||
)
|
||||
|
||||
output_parts = []
|
||||
if result.stdout.strip():
|
||||
output_parts.append(result.stdout.strip())
|
||||
if result.stderr.strip():
|
||||
output_parts.append(f"[stderr]\n{result.stderr.strip()}")
|
||||
if result.returncode != 0:
|
||||
output_parts.append(f"[exit code: {result.returncode}]")
|
||||
|
||||
output = "\n".join(output_parts) if output_parts else "(no output)"
|
||||
return _scrub_output(output, secrets)
|
||||
|
||||
except subprocess.TimeoutExpired:
|
||||
return f"Error: script timed out after {SCRIPT_TIMEOUT} seconds."
|
||||
finally:
|
||||
Path(script_path).unlink(missing_ok=True)
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def flows(name: str = "") -> str:
|
||||
"""Show flows: declarative multi-step work whose state lives in files.
|
||||
|
||||
Each step declares which files it needs and which it produces. Status is
|
||||
computed purely from the filesystem: blocked (a needed file is missing),
|
||||
ready (needs exist, produces missing), stale (a need changed after the
|
||||
produces were written), done. Instructions are shown only for actionable
|
||||
(ready or stale) steps.
|
||||
|
||||
You complete a step by writing its declared produces with write_context.
|
||||
Nothing else tracks progress; the files are the state.
|
||||
|
||||
Args:
|
||||
name: Optional flow name to show just one flow.
|
||||
"""
|
||||
all_flows = flows_mod.load_flows(PROJECT_DIR)
|
||||
if not all_flows:
|
||||
return "No flows defined in flows/*/flow.yaml"
|
||||
|
||||
if name:
|
||||
if name not in all_flows:
|
||||
return f"Error: no flow named {name}. Available: {', '.join(all_flows)}"
|
||||
all_flows = {name: all_flows[name]}
|
||||
|
||||
lines = []
|
||||
for flow in all_flows.values():
|
||||
lines.extend(flows_mod.render_flow(PROJECT_DIR, flow))
|
||||
lines.append("")
|
||||
return "\n".join(lines).rstrip()
|
||||
|
||||
|
||||
@mcp.tool
|
||||
def list_connections() -> str:
|
||||
"""Show all connections with their secrets, deps, context files, and whether each secret has a value."""
|
||||
connections = _load_connections()
|
||||
secrets = _load_secrets_env()
|
||||
|
||||
if not connections:
|
||||
return "No connections defined in connections/*/connection.yaml"
|
||||
|
||||
lines = []
|
||||
for cname, conn in connections.items():
|
||||
lines.append(f"## {cname}")
|
||||
if conn.description:
|
||||
lines.append(conn.description)
|
||||
lines.append("")
|
||||
lines.append("Secrets:")
|
||||
for s in conn.secrets:
|
||||
has_value = s in secrets and bool(secrets[s])
|
||||
icon = "filled" if has_value else "MISSING"
|
||||
lines.append(f" - {s}: {icon}")
|
||||
if conn.deps:
|
||||
lines.append(f"Deps: {', '.join(conn.deps)}")
|
||||
context_files = _connection_files(cname)
|
||||
if context_files:
|
||||
lines.append("Context:")
|
||||
for f in context_files:
|
||||
lines.append(f" - {f}")
|
||||
lines.append("")
|
||||
|
||||
return "\n".join(lines)
|
||||
@@ -0,0 +1,71 @@
|
||||
import os
|
||||
|
||||
import yaml
|
||||
|
||||
from gcontext.flows import flow_board, load_flows, step_state
|
||||
from gcontext.models import FlowStep
|
||||
|
||||
FLOW = {
|
||||
"name": "f",
|
||||
"steps": [
|
||||
{"id": "capture", "produces": ["brief.md"]},
|
||||
{"id": "draft", "needs": ["brief.md"], "produces": ["draft.md"]},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def make_flow(project, data=FLOW):
|
||||
d = project / "flows" / data["name"]
|
||||
d.mkdir(parents=True)
|
||||
(d / "flow.yaml").write_text(yaml.safe_dump(data))
|
||||
|
||||
|
||||
def test_load_flows(tmp_path):
|
||||
make_flow(tmp_path)
|
||||
flows = load_flows(tmp_path)
|
||||
assert list(flows) == ["f"]
|
||||
assert [s.id for s in flows["f"].steps] == ["capture", "draft"]
|
||||
|
||||
|
||||
def test_no_needs_is_ready(tmp_path):
|
||||
state = step_state(tmp_path, FlowStep(id="s", produces=["out.md"]))
|
||||
assert state["status"] == "ready"
|
||||
assert state["missing"] == ["out.md"]
|
||||
|
||||
|
||||
def test_blocked_then_ready_then_done(tmp_path):
|
||||
step = FlowStep(id="s", needs=["brief.md"], produces=["draft.md"])
|
||||
|
||||
state = step_state(tmp_path, step)
|
||||
assert state["status"] == "blocked"
|
||||
assert state["missing"] == ["brief.md"]
|
||||
|
||||
(tmp_path / "brief.md").write_text("brief")
|
||||
assert step_state(tmp_path, step)["status"] == "ready"
|
||||
|
||||
(tmp_path / "draft.md").write_text("draft")
|
||||
assert step_state(tmp_path, step)["status"] == "done"
|
||||
|
||||
|
||||
def test_stale_when_need_changes_after_produce(tmp_path):
|
||||
step = FlowStep(id="s", needs=["brief.md"], produces=["draft.md"])
|
||||
brief = tmp_path / "brief.md"
|
||||
draft = tmp_path / "draft.md"
|
||||
brief.write_text("brief")
|
||||
draft.write_text("draft")
|
||||
|
||||
now = draft.stat().st_mtime
|
||||
os.utime(brief, (now + 10, now + 10))
|
||||
|
||||
state = step_state(tmp_path, step)
|
||||
assert state["status"] == "stale"
|
||||
assert state["stale_needs"] == ["brief.md"]
|
||||
|
||||
|
||||
def test_board_statuses(tmp_path):
|
||||
make_flow(tmp_path)
|
||||
flow = load_flows(tmp_path)["f"]
|
||||
assert [s["status"] for s in flow_board(tmp_path, flow)] == ["ready", "blocked"]
|
||||
|
||||
(tmp_path / "brief.md").write_text("brief")
|
||||
assert [s["status"] for s in flow_board(tmp_path, flow)] == ["done", "ready"]
|
||||
@@ -0,0 +1,43 @@
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
|
||||
def run_cli(*args, cwd):
|
||||
return subprocess.run(
|
||||
[sys.executable, "-m", "gcontext.cli", *args],
|
||||
capture_output=True, text=True, cwd=cwd,
|
||||
)
|
||||
|
||||
|
||||
def test_init_scaffolds_agent(tmp_path):
|
||||
result = run_cli("init", "my-agent", cwd=tmp_path)
|
||||
assert result.returncode == 0, result.stderr
|
||||
|
||||
agent = tmp_path / "my-agent"
|
||||
for rel in [
|
||||
"gcontext.yaml",
|
||||
"instructions.md",
|
||||
"secrets.env",
|
||||
".gitignore",
|
||||
"connections/httpbin/connection.yaml",
|
||||
"flows/demo-brief/flow.yaml",
|
||||
]:
|
||||
assert (agent / rel).is_file(), rel
|
||||
assert "name: my-agent" in (agent / "gcontext.yaml").read_text()
|
||||
assert "secrets.env" in (agent / ".gitignore").read_text()
|
||||
|
||||
|
||||
def test_init_refuses_non_empty_dir(tmp_path):
|
||||
(tmp_path / "taken").mkdir()
|
||||
(tmp_path / "taken" / "x").write_text("x")
|
||||
result = run_cli("init", "taken", cwd=tmp_path)
|
||||
assert result.returncode == 1
|
||||
assert "not empty" in result.stderr
|
||||
|
||||
|
||||
def test_scaffolded_agent_works_with_cli(tmp_path):
|
||||
run_cli("init", "a", cwd=tmp_path)
|
||||
result = run_cli("flows", "a", cwd=tmp_path)
|
||||
assert result.returncode == 0, result.stderr
|
||||
assert "demo-brief" in result.stdout
|
||||
assert "capture" in result.stdout
|
||||
@@ -0,0 +1,69 @@
|
||||
import pytest
|
||||
|
||||
from gcontext import server
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def project(tmp_path, monkeypatch):
|
||||
(tmp_path / "gcontext.yaml").write_text("name: t\n")
|
||||
monkeypatch.setattr(server, "PROJECT_DIR", tmp_path)
|
||||
return tmp_path
|
||||
|
||||
|
||||
def test_scrub_output():
|
||||
secrets = {"API_KEY": "sk-verysecret", "SHORT": "ab"}
|
||||
out = server._scrub_output("token sk-verysecret used, ab kept", secrets)
|
||||
assert "sk-verysecret" not in out
|
||||
assert "***" in out
|
||||
assert "ab kept" in out # values of length <= 3 are not scrubbed
|
||||
|
||||
|
||||
def test_read_context_blocks_traversal(project):
|
||||
assert "outside the project" in server.read_context("../gcontext.yaml")
|
||||
assert "outside the project" in server.read_context("/etc/hosts")
|
||||
|
||||
|
||||
def test_write_context_blocks_traversal_and_protected_files(project):
|
||||
assert "outside the project" in server.write_context("../x.md", "hi")
|
||||
assert "Error" in server.write_context("secrets.env", "STOLEN=1")
|
||||
assert "Error" in server.write_context("connections/a/connection.yaml", "nope")
|
||||
|
||||
|
||||
def test_write_then_read_roundtrip(project):
|
||||
server.write_context("modules/notes/index.md", "hello")
|
||||
assert server.read_context("modules/notes/index.md") == "hello"
|
||||
|
||||
|
||||
def test_archive_not_scanned_but_reported(project):
|
||||
(project / "modules" / "active").mkdir(parents=True)
|
||||
(project / "modules" / "active" / "index.md").write_text("x")
|
||||
(project / "archive" / "modules" / "old").mkdir(parents=True)
|
||||
(project / "archive" / "modules" / "old" / "index.md").write_text("x")
|
||||
|
||||
modules = server._discover_modules()
|
||||
assert "active" in modules and "old" not in modules
|
||||
|
||||
assert server._archived() == {"modules": ["old"]}
|
||||
overview = server.overview()
|
||||
assert "## Archive" in overview
|
||||
assert "old" in overview
|
||||
|
||||
|
||||
def test_archive_readable_by_path(project):
|
||||
(project / "archive").mkdir()
|
||||
(project / "archive" / "note.md").write_text("kept")
|
||||
assert server.read_context("archive/note.md") == "kept"
|
||||
|
||||
|
||||
def test_flows_tool_and_ledger(project):
|
||||
d = project / "flows" / "f"
|
||||
d.mkdir(parents=True)
|
||||
(d / "flow.yaml").write_text(
|
||||
"name: f\nsteps:\n - id: s\n produces: [flows/f/out.md]\n instructions: write it\n"
|
||||
)
|
||||
out = server.flows()
|
||||
assert "[ready] s" in out
|
||||
assert "write it" in out # actionable steps expose instructions
|
||||
|
||||
g6 = [p for p in server.build_ledger("mcp") if p["id"] == "G6"]
|
||||
assert g6 and "1 flow(s)" in g6[0]["detail"]
|
||||
Reference in New Issue
Block a user