326 lines
8.7 KiB
Python
326 lines
8.7 KiB
Python
"""Treasury MCP tools - Budget management operations."""
|
|
|
|
import json
|
|
import os
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
|
|
import blake3
|
|
|
|
# VaultMesh root from env or default
|
|
VAULTMESH_ROOT = Path(os.environ.get("VAULTMESH_ROOT", Path(__file__).parents[3])).resolve()
|
|
TREASURY_JSONL = VAULTMESH_ROOT / "receipts/treasury/treasury_events.jsonl"
|
|
TREASURY_STATE = VAULTMESH_ROOT / "receipts/treasury/budgets.json"
|
|
|
|
# Schema version
|
|
SCHEMA_VERSION = "2.0.0"
|
|
|
|
|
|
def _vmhash_blake3(data: bytes) -> str:
|
|
"""VaultMesh hash: blake3:<hex>."""
|
|
return f"blake3:{blake3.blake3(data).hexdigest()}"
|
|
|
|
|
|
def _load_budgets() -> dict[str, dict]:
|
|
"""Load current budget state from disk."""
|
|
if not TREASURY_STATE.exists():
|
|
return {}
|
|
try:
|
|
return json.loads(TREASURY_STATE.read_text())
|
|
except (json.JSONDecodeError, FileNotFoundError):
|
|
return {}
|
|
|
|
|
|
def _save_budgets(budgets: dict[str, dict]) -> None:
|
|
"""Persist budget state to disk."""
|
|
TREASURY_STATE.parent.mkdir(parents=True, exist_ok=True)
|
|
TREASURY_STATE.write_text(json.dumps(budgets, indent=2))
|
|
|
|
|
|
def _emit_receipt(receipt_type: str, body: dict, tags: list[str]) -> dict:
|
|
"""Emit a treasury receipt to JSONL."""
|
|
body_json = json.dumps(body, sort_keys=True)
|
|
root_hash = _vmhash_blake3(body_json.encode())
|
|
|
|
receipt = {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"type": receipt_type,
|
|
"timestamp": datetime.now(timezone.utc).isoformat(),
|
|
"scroll": "treasury",
|
|
"tags": tags,
|
|
"root_hash": root_hash,
|
|
"body": body,
|
|
}
|
|
|
|
TREASURY_JSONL.parent.mkdir(parents=True, exist_ok=True)
|
|
with open(TREASURY_JSONL, "a") as f:
|
|
f.write(json.dumps(receipt) + "\n")
|
|
|
|
# Update ROOT file
|
|
_update_root()
|
|
|
|
return receipt
|
|
|
|
|
|
def _update_root() -> None:
|
|
"""Update ROOT.treasury.txt with current Merkle root."""
|
|
if not TREASURY_JSONL.exists():
|
|
return
|
|
|
|
hashes = []
|
|
with open(TREASURY_JSONL, "r") as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if line:
|
|
hashes.append(_vmhash_blake3(line.encode()))
|
|
|
|
if not hashes:
|
|
root = _vmhash_blake3(b"empty")
|
|
elif len(hashes) == 1:
|
|
root = hashes[0]
|
|
else:
|
|
current = hashes
|
|
while len(current) > 1:
|
|
next_level = []
|
|
for i in range(0, len(current), 2):
|
|
if i + 1 < len(current):
|
|
combined = current[i] + current[i + 1]
|
|
else:
|
|
combined = current[i] + current[i]
|
|
next_level.append(_vmhash_blake3(combined.encode()))
|
|
current = next_level
|
|
root = current[0]
|
|
|
|
root_file = VAULTMESH_ROOT / "ROOT.treasury.txt"
|
|
root_file.write_text(root)
|
|
|
|
|
|
def treasury_create_budget(
|
|
budget_id: str,
|
|
name: str,
|
|
allocated: int,
|
|
currency: str = "EUR",
|
|
created_by: str = "did:vm:mcp:treasury",
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Create a new budget.
|
|
|
|
Args:
|
|
budget_id: Unique identifier for the budget
|
|
name: Human-readable budget name
|
|
allocated: Initial allocation amount (cents/smallest unit)
|
|
currency: Currency code (default: EUR)
|
|
created_by: DID of the actor creating the budget
|
|
|
|
Returns:
|
|
Created budget with receipt info
|
|
"""
|
|
budgets = _load_budgets()
|
|
|
|
if budget_id in budgets:
|
|
return {"error": f"Budget already exists: {budget_id}"}
|
|
|
|
now = datetime.now(timezone.utc)
|
|
budget = {
|
|
"id": budget_id,
|
|
"name": name,
|
|
"currency": currency,
|
|
"allocated": allocated,
|
|
"spent": 0,
|
|
"created_at": now.isoformat(),
|
|
"created_by": created_by,
|
|
}
|
|
|
|
budgets[budget_id] = budget
|
|
_save_budgets(budgets)
|
|
|
|
# Emit receipt
|
|
receipt_body = {
|
|
"budget_id": budget_id,
|
|
"name": name,
|
|
"currency": currency,
|
|
"allocated": allocated,
|
|
"created_by": created_by,
|
|
}
|
|
receipt = _emit_receipt(
|
|
"treasury_budget_create",
|
|
receipt_body,
|
|
["treasury", "budget", "create", budget_id],
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"budget": budget,
|
|
"receipt_hash": receipt["root_hash"],
|
|
"message": f"Created budget '{name}' with {allocated} {currency}",
|
|
}
|
|
|
|
|
|
def treasury_debit(
|
|
budget_id: str,
|
|
amount: int,
|
|
description: str,
|
|
debited_by: str = "did:vm:mcp:treasury",
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Debit (spend) from a budget.
|
|
|
|
Args:
|
|
budget_id: Budget to debit from
|
|
amount: Amount to debit (cents/smallest unit)
|
|
description: Description of the expenditure
|
|
debited_by: DID of the actor making the debit
|
|
|
|
Returns:
|
|
Updated budget with receipt info
|
|
"""
|
|
budgets = _load_budgets()
|
|
|
|
if budget_id not in budgets:
|
|
return {"error": f"Budget not found: {budget_id}"}
|
|
|
|
budget = budgets[budget_id]
|
|
remaining = budget["allocated"] - budget["spent"]
|
|
|
|
if amount > remaining:
|
|
return {
|
|
"error": "Insufficient funds",
|
|
"budget_id": budget_id,
|
|
"requested": amount,
|
|
"available": remaining,
|
|
}
|
|
|
|
budget["spent"] += amount
|
|
budgets[budget_id] = budget
|
|
_save_budgets(budgets)
|
|
|
|
# Emit receipt
|
|
receipt_body = {
|
|
"budget_id": budget_id,
|
|
"amount": amount,
|
|
"currency": budget["currency"],
|
|
"description": description,
|
|
"debited_by": debited_by,
|
|
"new_spent": budget["spent"],
|
|
"new_remaining": budget["allocated"] - budget["spent"],
|
|
}
|
|
receipt = _emit_receipt(
|
|
"treasury_debit",
|
|
receipt_body,
|
|
["treasury", "debit", budget_id],
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"budget": budget,
|
|
"remaining": budget["allocated"] - budget["spent"],
|
|
"receipt_hash": receipt["root_hash"],
|
|
"message": f"Debited {amount} from '{budget['name']}' - {description}",
|
|
}
|
|
|
|
|
|
def treasury_credit(
|
|
budget_id: str,
|
|
amount: int,
|
|
description: str,
|
|
credited_by: str = "did:vm:mcp:treasury",
|
|
) -> dict[str, Any]:
|
|
"""
|
|
Credit (add funds) to a budget.
|
|
|
|
Args:
|
|
budget_id: Budget to credit
|
|
amount: Amount to add (cents/smallest unit)
|
|
description: Description of the credit (refund, adjustment, etc.)
|
|
credited_by: DID of the actor making the credit
|
|
|
|
Returns:
|
|
Updated budget with receipt info
|
|
"""
|
|
budgets = _load_budgets()
|
|
|
|
if budget_id not in budgets:
|
|
return {"error": f"Budget not found: {budget_id}"}
|
|
|
|
budget = budgets[budget_id]
|
|
budget["allocated"] += amount
|
|
budgets[budget_id] = budget
|
|
_save_budgets(budgets)
|
|
|
|
# Emit receipt
|
|
receipt_body = {
|
|
"budget_id": budget_id,
|
|
"amount": amount,
|
|
"currency": budget["currency"],
|
|
"description": description,
|
|
"credited_by": credited_by,
|
|
"new_allocated": budget["allocated"],
|
|
}
|
|
receipt = _emit_receipt(
|
|
"treasury_credit",
|
|
receipt_body,
|
|
["treasury", "credit", budget_id],
|
|
)
|
|
|
|
return {
|
|
"success": True,
|
|
"budget": budget,
|
|
"remaining": budget["allocated"] - budget["spent"],
|
|
"receipt_hash": receipt["root_hash"],
|
|
"message": f"Credited {amount} to '{budget['name']}' - {description}",
|
|
}
|
|
|
|
|
|
def treasury_balance(budget_id: Optional[str] = None) -> dict[str, Any]:
|
|
"""
|
|
Get budget balance(s).
|
|
|
|
Args:
|
|
budget_id: Specific budget ID (optional, returns all if omitted)
|
|
|
|
Returns:
|
|
Budget balance(s) with current state
|
|
"""
|
|
budgets = _load_budgets()
|
|
|
|
if budget_id:
|
|
if budget_id not in budgets:
|
|
return {"error": f"Budget not found: {budget_id}"}
|
|
budget = budgets[budget_id]
|
|
return {
|
|
"budget_id": budget_id,
|
|
"name": budget["name"],
|
|
"currency": budget["currency"],
|
|
"allocated": budget["allocated"],
|
|
"spent": budget["spent"],
|
|
"remaining": budget["allocated"] - budget["spent"],
|
|
}
|
|
|
|
# Return all budgets
|
|
result = []
|
|
total_allocated = 0
|
|
total_spent = 0
|
|
for bid, budget in budgets.items():
|
|
remaining = budget["allocated"] - budget["spent"]
|
|
total_allocated += budget["allocated"]
|
|
total_spent += budget["spent"]
|
|
result.append({
|
|
"budget_id": bid,
|
|
"name": budget["name"],
|
|
"currency": budget["currency"],
|
|
"allocated": budget["allocated"],
|
|
"spent": budget["spent"],
|
|
"remaining": remaining,
|
|
})
|
|
|
|
return {
|
|
"budgets": result,
|
|
"count": len(result),
|
|
"totals": {
|
|
"allocated": total_allocated,
|
|
"spent": total_spent,
|
|
"remaining": total_allocated - total_spent,
|
|
},
|
|
}
|