orchestrata_ultimate.gateway.mcp_approval_gate

Reference › orchestrata_ultimate › gateway › mcp_approval_gate

McpApprovalGate issues and redeems single-use approval tokens for the MCP transport (SPEC-014 REQ-014-205, ADR-0099). mcp_transport uses it to guard every tool that mutates state. issue mints a random token bound to one capability, the tool name, with an expiry set by the constructor's time-to-live. redeem returns True and deletes the token only if it exists, was issued for the same capability and has not expired; any other case returns False, so a token cannot be reused or applied to a different tool. A lock makes both calls thread-safe. Tokens are kept in memory only.

At a glance

Class diagram

Dependency diagram

API

Classes

McpApprovalGate

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

Issues and redeems single-use approval tokens.

Source

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

"""McpApprovalGate -- single-use, tool-bound approval tokens.

ADR-0038 rule 2 / ADR-0099 rule 3, the same contract codetools ships: a
mutating tool call needs a token this process issued. The client first calls
the ``approve`` tool (the host's permission prompt for that tool is where a
human consents) and receives an unguessable token bound to ONE tool, valid
once, for 300 seconds. A token the server did not issue, one bound to another
tool, an expired one, or one already redeemed is refused before any side
effect. Tokens are never logged, printed or written to the ledger.

Author: Mohamed
"""

# Spec: 014 - WebSocket Gateway

from __future__ import annotations

import secrets
import threading
import time

from orchestrata_ultimate.core.base.traces import traces

DEFAULT_TTL_SECONDS = 300.0


@traces(spec="014", story="US-013", reqs=["REQ-014-205"])
class McpApprovalGate:
    """Issues and redeems single-use approval tokens."""

    def __init__(self, ttl_seconds: float = DEFAULT_TTL_SECONDS) -> None:
        self._ttl = ttl_seconds
        self._issued: dict[str, tuple[str, float]] = {}
        self._lock = threading.Lock()

    def issue(self, capability: str) -> str:
        """Mint a token bound to ``capability``."""
        token = secrets.token_urlsafe(24)
        with self._lock:
            self._issued[token] = (capability, time.monotonic() + self._ttl)
        return token

    def redeem(self, token: str | None, capability: str) -> bool:
        """Consume ``token`` for ``capability``; ``False`` when it is refused."""
        if not token:
            return False
        with self._lock:
            entry = self._issued.get(token)
            valid = (
                entry is not None
                and entry[0] == capability
                and time.monotonic() <= entry[1]
            )
            if valid:
                del self._issued[token]  # single use
        return valid