orchestrata_ultimate.gateway.mcp_ledger

Reference › orchestrata_ultimate › gateway › mcp_ledger

McpLedger is an append-only ndjson record of MCP tool-call outcomes (SPEC-014 REQ-014-205), written by mcp_transport for each approval, refusal, failure and success. default_path returns the path in the ORCHESTRATA_MCP_LEDGER variable, or else mcp-ledger.ndjson under the user's .local/state/orchestrata directory; the environment and home directory can be passed in so tests never touch the real ones. append_row writes one JSON line with a short id, a timestamp, the tool, the outcome and an exit code, creating parent directories as needed under a lock, and returns the row id, which the transport includes in its refusal and failure replies.

At a glance

Class diagram

Dependency diagram

API

Classes

McpLedger

Decorators: traces(spec='014', story='US-013', reqs=['REQ-014-205']) · Line 31

Append-only ndjson ledger of MCP tool-call outcomes.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/mcp_ledger.py

"""McpLedger -- append-only ndjson ledger of MCP tool calls.

ADR-0038 rule 2 / ADR-0099: one row per call outcome (``ok``, ``refused``,
``failed``, ``approved``), keyed by a short id that a refusal carries back to
the client so it can be traced to its row. Rows hold the tool name, outcome
and exit code only -- never call arguments, never an approval token, never
the gateway token. The path is ``$ORCHESTRATA_MCP_LEDGER``, else
``~/.local/state/orchestrata/mcp-ledger.ndjson``.

Author: Mohamed
"""

# Spec: 014 - WebSocket Gateway

from __future__ import annotations

import json
import os
import threading
import time
import uuid
from collections.abc import Mapping
from pathlib import Path

from orchestrata_ultimate.core.base.traces import traces

LEDGER_ENV = "ORCHESTRATA_MCP_LEDGER"


@traces(spec="014", story="US-013", reqs=["REQ-014-205"])
class McpLedger:
    """Append-only ndjson ledger of MCP tool-call outcomes."""

    def __init__(self, path: Path | None = None) -> None:
        self._path = path if path is not None else self.default_path()
        self._lock = threading.Lock()

    @staticmethod
    def default_path(
        environ: Mapping[str, str] | None = None, home: Path | None = None
    ) -> Path:
        """``$ORCHESTRATA_MCP_LEDGER``, else the per-user state path.

        ``environ`` and ``home`` are explicit seams (default: the process
        environment and ``Path.home()``) so callers and tests never mutate
        the real environment.
        """
        env = os.environ if environ is None else environ
        override = env.get(LEDGER_ENV)
        if override:
            return Path(override)
        base = home if home is not None else Path.home()
        return base / ".local" / "state" / "orchestrata" / "mcp-ledger.ndjson"

    @property
    def path(self) -> Path:
        """Where rows are appended."""
        return self._path

    def append_row(self, tool: str, outcome: str, exit_code: int = 0) -> str:
        """Append one row and return its id."""
        row_id = uuid.uuid4().hex[:12]
        row = {
            "id": row_id,
            "ts": time.time(),
            "tool": tool,
            "outcome": outcome,
            "exit_code": exit_code,
        }
        with self._lock:
            self._path.parent.mkdir(parents=True, exist_ok=True)
            with self._path.open("a", encoding="utf-8") as handle:
                handle.write(json.dumps(row, sort_keys=True) + "\n")
        return row_id