feat: hash-only stamp, verify, and hash_report MCP tools

Memory ledger for stdio tests; NatsLedger speaks locald subjects.

README.md

@@ -2,28 +2,36 @@
 
 [![standard-readme compliant](https://img.shields.io/badge/readme%20style-standard-brightgreen.svg?style=flat-square)](https://github.com/RichardLitt/standard-readme)
 
-Verae MCP timestamping server. Implementation starts after the research plan is approved.
+Verae MCP timestamping server. Hash-only stamps via leaf locald/NATS.
 
-The plan and MCP requirements live in [verae-mcp-research](../verae-mcp-research). This repo will be the Streamable HTTP resource server: tools to create and verify timestamps, OAuth bearer verification, no login UI of its own.
+The MCP process never hashes content. Clients send a SHA-256 hex digest. Tools `stamp`, `verify`, and `hash_report` use an in-memory ledger over stdio, or `NatsLedger` against `verae.ts.request` / `verae.ts.verify` / `verae.hash.report` (leaf `locald`). Plan and product model: [verae-mcp-research](../verae-mcp-research).
 
 ## Install
 
 ```sh
 cd /Users/marchon/research/modelcontextprotocol/verae-mcp-timestamping-server
-# Python SDK 2+ after scaffolding: uv add "mcp[cli]"
+uv sync --group dev
 ```
 
 ## Usage
 
+Inspector (stdio, memory ledger):
+
 ```sh
-# After implementation, Inspector over stdio:
-# uv run mcp dev src/verae_mcp_timestamping_server/server.py
-echo "not implemented yet; see ../verae-mcp-research/docs/superpowers/plans/2026-09-20-verae-mcp-timestamping.md"
+uv run mcp dev src/verae_mcp_timestamping_server/server.py
 ```
 
+Tests:
+
+```sh
+uv run pytest
+```
+
+Point a later HTTP deploy at a leaf with `NatsLedger` and `VERAE_NATS_URL=nats://verae-nats-bus:4222`. Do not send file bytes to `stamp`.
+
 ## Contributing
 
-Questions: this research workspace. PRs welcome once the plan in `verae-mcp-research` is agreed. Protocol questions belong at https://github.com/modelcontextprotocol.
+Questions: this research workspace. PRs welcome. Protocol questions belong at https://github.com/modelcontextprotocol. Timestamp ledger questions belong with `verae-timestamp` / `locald`.
 
 ## License
 

pyproject.toml

@@ -0,0 +1,32 @@
+[project]
+name = "verae-mcp-timestamping-server"
+version = "0.1.0"
+description = "Verae MCP timestamping server. Hash-only stamps via leaf locald/NATS."
+readme = "README.md"
+requires-python = ">=3.10"
+license = { text = "UNLICENSED" }
+authors = [{ name = "George Lambert", email = "marchon@gmail.com" }]
+dependencies = [
+    "mcp[cli]>=2.0.0",
+    "nats-py>=2.9.0",
+    "pydantic>=2.0.0",
+]
+
+[project.scripts]
+verae-mcp-timestamping-server = "verae_mcp_timestamping_server.server:main"
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["src/verae_mcp_timestamping_server"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[dependency-groups]
+dev = [
+    "pytest>=8.4.0",
+    "anyio>=4.0.0",
+]

src/verae_mcp_timestamping_server/__init__.py

@@ -0,0 +1,3 @@
+"""Verae MCP timestamping server."""
+
+__version__ = "0.1.0"

src/verae_mcp_timestamping_server/ledger.py

@@ -0,0 +1,146 @@
+"""Hash-only timestamp ledger. Memory implementation for tests and stdio."""
+
+from __future__ import annotations
+
+import hashlib
+import re
+import time
+from typing import Protocol
+
+from pydantic import BaseModel
+
+SHA256_EMPTY = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+_SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+_SCOPE_RANK = {"local": 1, "shared": 2, "global": 3}
+
+
+class Receipt(BaseModel):
+    receipt_id: str
+    hash: str
+    scope: str
+    cube_id: str = ""
+    org_id: str = ""
+    created_at: str
+    merkle: str
+    restamp_of: str = ""
+
+
+class Receipts(BaseModel):
+    receipts: list[Receipt]
+
+
+class VerifyResult(BaseModel):
+    valid: bool
+    error: str | None = None
+    receipt: Receipt | None = None
+
+
+class Ledger(Protocol):
+    async def stamp(
+        self,
+        sha256: str,
+        scope: str = "local",
+        cube_id: str = "",
+        org_id: str = "",
+    ) -> list[Receipt]: ...
+
+    async def verify(self, receipt_id: str, want_scope: str = "") -> VerifyResult: ...
+
+    async def hash_report(
+        self, sha256: str, org_id: str = "", want_scope: str = ""
+    ) -> list[Receipt]: ...
+
+
+def validate_sha256(value: str) -> str:
+    digest = value.strip().lower()
+    if digest.startswith("sha256:"):
+        digest = digest[7:]
+    if not _SHA256_RE.match(digest):
+        raise ValueError("sha256 must be 64 lowercase hex characters")
+    return digest
+
+
+def _merkle(ids: list[str]) -> str:
+    if not ids:
+        return "sha256:" + SHA256_EMPTY
+    level = [hashlib.sha256(i.encode()).digest() for i in sorted(ids)]
+    while len(level) > 1:
+        nxt: list[bytes] = []
+        for i in range(0, len(level), 2):
+            if i + 1 == len(level):
+                nxt.append(level[i])
+            else:
+                nxt.append(hashlib.sha256(level[i] + level[i + 1]).digest())
+        level = nxt
+    return "sha256:" + level[0].hex()
+
+
+class MemoryLedger:
+    def __init__(self) -> None:
+        self._by_id: dict[str, Receipt] = {}
+        self._order: list[str] = []
+
+    def _key(self, sha256: str, org_id: str, scope: str) -> tuple[str, str, str]:
+        return (sha256, org_id, scope)
+
+    async def stamp(
+        self,
+        sha256: str,
+        scope: str = "local",
+        cube_id: str = "",
+        org_id: str = "",
+    ) -> list[Receipt]:
+        digest = validate_sha256(sha256)
+        if scope not in _SCOPE_RANK:
+            raise ValueError("scope must be local, shared, or global")
+        for rec in self._by_id.values():
+            if rec.hash == digest and rec.org_id == org_id and rec.scope == scope:
+                return [rec]
+        created = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        receipt_id = hashlib.sha256(
+            f"{digest}|{created}|{scope}|{org_id}".encode()
+        ).hexdigest()
+        rec = Receipt(
+            receipt_id=receipt_id,
+            hash=digest,
+            scope=scope,
+            cube_id=cube_id,
+            org_id=org_id,
+            created_at=created,
+            merkle="",
+        )
+        rec.merkle = _merkle([*self._order, rec.receipt_id])
+        self._by_id[rec.receipt_id] = rec
+        self._order.append(rec.receipt_id)
+        return [rec]
+
+    async def verify(self, receipt_id: str, want_scope: str = "") -> VerifyResult:
+        rec = self._by_id.get(receipt_id)
+        if rec is None:
+            return VerifyResult(valid=False, error="unknown receipt")
+        if want_scope:
+            if _SCOPE_RANK.get(rec.scope, 0) < _SCOPE_RANK.get(want_scope, 99):
+                return VerifyResult(
+                    valid=False,
+                    error=f"receipt scope {rec.scope} cannot satisfy {want_scope}",
+                    receipt=rec,
+                )
+        return VerifyResult(valid=True, receipt=rec)
+
+    async def hash_report(
+        self, sha256: str, org_id: str = "", want_scope: str = ""
+    ) -> list[Receipt]:
+        digest = validate_sha256(sha256)
+        out: list[Receipt] = []
+        for rid in self._order:
+            rec = self._by_id[rid]
+            if rec.hash != digest:
+                continue
+            if org_id and rec.org_id != org_id:
+                continue
+            if want_scope and _SCOPE_RANK.get(rec.scope, 0) < _SCOPE_RANK.get(
+                want_scope, 99
+            ):
+                continue
+            out.append(rec)
+        return out

src/verae_mcp_timestamping_server/nats_ledger.py

@@ -0,0 +1,87 @@
+"""Ledger over NATS subjects served by verae-timestamp locald."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from verae_mcp_timestamping_server.ledger import (
+    Receipt,
+    VerifyResult,
+    validate_sha256,
+)
+
+SUBJECT_REQUEST = "verae.ts.request"
+SUBJECT_VERIFY = "verae.ts.verify"
+SUBJECT_REPORT = "verae.hash.report"
+
+
+class NatsLedger:
+    def __init__(self, nc: Any, timeout: float = 2.0) -> None:
+        self._nc = nc
+        self._timeout = timeout
+
+    async def _ask(self, subject: str, payload: dict[str, Any]) -> dict[str, Any] | list[Any]:
+        msg = await self._nc.request(
+            subject, json.dumps(payload).encode(), timeout=self._timeout
+        )
+        return json.loads(msg.data)
+
+    async def stamp(
+        self,
+        sha256: str,
+        scope: str = "local",
+        cube_id: str = "",
+        org_id: str = "",
+    ) -> list[Receipt]:
+        digest = validate_sha256(sha256)
+        body = await self._ask(
+            SUBJECT_REQUEST,
+            {
+                "hash": digest,
+                "scope": scope,
+                "cube_id": cube_id,
+                "org_id": org_id,
+            },
+        )
+        if isinstance(body, dict) and body.get("error"):
+            raise ValueError(str(body["error"]))
+        if isinstance(body, list):
+            return [Receipt.model_validate(item) for item in body]
+        return [Receipt.model_validate(body)]
+
+    async def verify(self, receipt_id: str, want_scope: str = "") -> VerifyResult:
+        body = await self._ask(
+            SUBJECT_VERIFY,
+            {"receipt_id": receipt_id, "want_scope": want_scope},
+        )
+        if not isinstance(body, dict):
+            return VerifyResult(valid=False, error="invalid verify payload")
+        if body.get("valid") is False or body.get("error"):
+            return VerifyResult(valid=False, error=body.get("error"))
+        if "receipt_id" in body:
+            return VerifyResult(valid=True, receipt=Receipt.model_validate(body))
+        return VerifyResult(valid=False, error="unknown receipt")
+
+    async def hash_report(
+        self, sha256: str, org_id: str = "", want_scope: str = ""
+    ) -> list[Receipt]:
+        digest = validate_sha256(sha256)
+        body = await self._ask(
+            SUBJECT_REPORT,
+            {"hash": digest, "org_id": org_id, "want_scope": want_scope},
+        )
+        if isinstance(body, list):
+            return [Receipt.model_validate(item) for item in body]
+        if not isinstance(body, dict):
+            return []
+        if body.get("error"):
+            raise ValueError(str(body["error"]))
+        if "receipts" in body:
+            return [Receipt.model_validate(item) for item in body["receipts"]]
+        out: list[Receipt] = []
+        for event in body.get("timeline") or []:
+            rec = event.get("receipt") if isinstance(event, dict) else None
+            if rec:
+                out.append(Receipt.model_validate(rec))
+        return out

src/verae_mcp_timestamping_server/server.py

@@ -0,0 +1,77 @@
+"""MCP resource server: stamp / verify / hash_report. Hex only, never content."""
+
+from __future__ import annotations
+
+import os
+
+from mcp.server import MCPServer
+from mcp.server.mcpserver.exceptions import ToolError
+
+from verae_mcp_timestamping_server.ledger import (
+    Ledger,
+    MemoryLedger,
+    Receipts,
+    VerifyResult,
+)
+
+mcp = MCPServer("verae-timestamping")
+_ledger: Ledger = MemoryLedger()
+
+
+def build_mcp(ledger: Ledger | None = None) -> MCPServer:
+    global _ledger
+    _ledger = ledger or MemoryLedger()
+    return mcp
+
+
+def _tool_error(exc: ValueError) -> ToolError:
+    return ToolError(str(exc))
+
+
+@mcp.tool()
+async def stamp(
+    sha256: str, scope: str = "local", cube_id: str = "", org_id: str = ""
+) -> Receipts:
+    """Commit a client-computed SHA-256 to the Verae leaf ledger. Hex only."""
+    try:
+        receipts = await _ledger.stamp(
+            sha256, scope=scope, cube_id=cube_id, org_id=org_id
+        )
+    except ValueError as exc:
+        raise _tool_error(exc) from exc
+    return Receipts(receipts=receipts)
+
+
+@mcp.tool()
+async def verify(receipt_id: str, want_scope: str = "") -> VerifyResult:
+    """Check a timestamp receipt. want_scope=global rejects a weaker local receipt."""
+    return await _ledger.verify(receipt_id, want_scope=want_scope)
+
+
+@mcp.tool()
+async def hash_report(
+    sha256: str, org_id: str = "", want_scope: str = ""
+) -> Receipts:
+    """JSON-array history of receipts for one SHA-256 (Carfax)."""
+    try:
+        receipts = await _ledger.hash_report(
+            sha256, org_id=org_id, want_scope=want_scope
+        )
+    except ValueError as exc:
+        raise _tool_error(exc) from exc
+    return Receipts(receipts=receipts)
+
+
+def main() -> None:
+    url = os.environ.get("VERAE_NATS_URL")
+    if url:
+        raise SystemExit(
+            "NATS leaf mode is wired through NatsLedger; "
+            "stdio Task 1 uses MemoryLedger. Set VERAE_NATS_URL on Streamable HTTP later."
+        )
+    build_mcp(MemoryLedger())
+    mcp.run()
+
+
+if __name__ == "__main__":
+    main()

tests/conftest.py

@@ -0,0 +1,6 @@
+import pytest
+
+
+@pytest.fixture
+def anyio_backend():
+    return "asyncio"

tests/test_ledger.py

@@ -0,0 +1,54 @@
+"""Memory ledger: hash-only stamp, verify, report. Same hash is idempotent."""
+
+import pytest
+
+from verae_mcp_timestamping_server.ledger import MemoryLedger, SHA256_EMPTY
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_stamp_returns_json_array_with_one_receipt():
+    ledger = MemoryLedger()
+    receipts = await ledger.stamp(SHA256_EMPTY, scope="local")
+    assert isinstance(receipts, list)
+    assert len(receipts) == 1
+    rec = receipts[0]
+    assert rec.hash == SHA256_EMPTY
+    assert rec.scope == "local"
+    assert rec.receipt_id
+    assert rec.created_at
+    assert rec.merkle
+
+
+async def test_same_hash_returns_same_receipt():
+    ledger = MemoryLedger()
+    first = await ledger.stamp(SHA256_EMPTY, scope="local")
+    second = await ledger.stamp(SHA256_EMPTY, scope="local")
+    assert first[0].receipt_id == second[0].receipt_id
+    assert first[0].created_at == second[0].created_at
+
+
+async def test_rejects_non_sha256():
+    ledger = MemoryLedger()
+    with pytest.raises(ValueError, match="sha256"):
+        await ledger.stamp("not-a-hash", scope="local")
+
+
+async def test_verify_receipt_and_scope_rank():
+    ledger = MemoryLedger()
+    [rec] = await ledger.stamp(SHA256_EMPTY, scope="local")
+    ok = await ledger.verify(rec.receipt_id, want_scope="local")
+    assert ok.valid is True
+    assert ok.receipt is not None
+    assert ok.receipt.receipt_id == rec.receipt_id
+
+    weak = await ledger.verify(rec.receipt_id, want_scope="global")
+    assert weak.valid is False
+
+
+async def test_hash_report_is_json_array():
+    ledger = MemoryLedger()
+    await ledger.stamp(SHA256_EMPTY, scope="local")
+    report = await ledger.hash_report(SHA256_EMPTY)
+    assert isinstance(report, list)
+    assert report[0].hash == SHA256_EMPTY

tests/test_nats_ledger.py

@@ -0,0 +1,64 @@
+"""NatsLedger speaks locald subjects: verae.ts.request / verify / hash.report."""
+
+import json
+
+import pytest
+
+from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger, Receipt
+from verae_mcp_timestamping_server.nats_ledger import NatsLedger
+
+pytestmark = pytest.mark.anyio
+
+
+class _Msg:
+    def __init__(self, data: bytes):
+        self.data = data
+
+
+class _FakeNc:
+    """Request/reply shaped like nats-py, backed by MemoryLedger (locald protocol)."""
+
+    def __init__(self, inner: MemoryLedger):
+        self.inner = inner
+
+    async def request(self, subject: str, payload: bytes, timeout: float = 2.0):
+        body = json.loads(payload)
+        if subject == "verae.ts.request":
+            receipts = await self.inner.stamp(
+                body["hash"],
+                scope=body.get("scope") or "local",
+                cube_id=body.get("cube_id") or "",
+                org_id=body.get("org_id") or "",
+            )
+            return _Msg(receipts[0].model_dump_json().encode())
+        if subject == "verae.ts.verify":
+            result = await self.inner.verify(
+                body["receipt_id"], want_scope=body.get("want_scope") or ""
+            )
+            if not result.valid or result.receipt is None:
+                return _Msg(json.dumps({"valid": False, "error": result.error}).encode())
+            return _Msg(result.receipt.model_dump_json().encode())
+        if subject == "verae.hash.report":
+            receipts = await self.inner.hash_report(
+                body["hash"], org_id=body.get("org_id") or ""
+            )
+            return _Msg(
+                json.dumps({"receipts": [r.model_dump() for r in receipts]}).encode()
+            )
+        raise AssertionError(f"unexpected subject {subject}")
+
+
+async def test_nats_stamp_verify_report_roundtrip():
+    nc = _FakeNc(MemoryLedger())
+    ledger = NatsLedger(nc)
+    receipts = await ledger.stamp(SHA256_EMPTY, scope="local", cube_id="cube-1")
+    assert receipts[0].hash == SHA256_EMPTY
+    assert receipts[0].cube_id == "cube-1"
+
+    ok = await ledger.verify(receipts[0].receipt_id, want_scope="local")
+    assert ok.valid is True
+    assert ok.receipt is not None
+
+    report = await ledger.hash_report(SHA256_EMPTY)
+    assert [r.receipt_id for r in report] == [receipts[0].receipt_id]
+    assert isinstance(report[0], Receipt)

tests/test_server.py

@@ -0,0 +1,54 @@
+"""In-memory MCP Client against hash-only tools."""
+
+import pytest
+from mcp import Client
+
+from verae_mcp_timestamping_server.ledger import SHA256_EMPTY, MemoryLedger
+from verae_mcp_timestamping_server.server import build_mcp
+
+pytestmark = pytest.mark.anyio
+
+
+async def test_stamp_tool_returns_receipts():
+    mcp = build_mcp(MemoryLedger())
+    async with Client(mcp, raise_exceptions=True) as client:
+        result = await client.call_tool(
+            "stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
+        )
+        assert result.is_error is not True
+        data = result.structured_content
+        assert data is not None
+        receipts = data["receipts"]
+        assert receipts[0]["hash"] == SHA256_EMPTY
+        assert receipts[0]["scope"] == "local"
+
+
+async def test_stamp_does_not_accept_content_bytes():
+    mcp = build_mcp(MemoryLedger())
+    async with Client(mcp, raise_exceptions=True) as client:
+        listed = await client.list_tools()
+        stamp = next(t for t in listed.tools if t.name == "stamp")
+        props = stamp.input_schema["properties"]
+        assert "sha256" in props
+        assert "content" not in props
+        assert "bytes" not in props
+        assert "file" not in props
+
+
+async def test_verify_and_hash_report_tools():
+    mcp = build_mcp(MemoryLedger())
+    async with Client(mcp, raise_exceptions=True) as client:
+        stamped = await client.call_tool(
+            "stamp", {"sha256": SHA256_EMPTY, "scope": "local"}
+        )
+        receipts = stamped.structured_content["receipts"]
+        receipt_id = receipts[0]["receipt_id"]
+
+        verified = await client.call_tool(
+            "verify", {"receipt_id": receipt_id, "want_scope": "local"}
+        )
+        assert verified.structured_content["valid"] is True
+
+        report = await client.call_tool("hash_report", {"sha256": SHA256_EMPTY})
+        items = report.structured_content["receipts"]
+        assert items[0]["receipt_id"] == receipt_id