RFC-010: Autonomous Intelligence Governance Protocol (AIGP) — 8. Python Client Implementation

AIGP Specification › RFC-010: Autonomous Intelligence Governance Protocol (AIGP) › 8. Python Client Implementation

← 7. Registration Schema · Section index · 9. Governance Authority (governance-server) Schema →

8. Python Client Implementation

class GovernanceClient:
    """AIGP client  consent-based autonomous intelligence governance protocol."""

    def __init__(self, gov_url: str, app_id: str, hmac_secret: str,
                 enabled: bool = True, mode: str = "REPORT"):
        self._url = gov_url.rstrip("/")
        self._app_id = app_id
        self._secret = hmac_secret
        self._enabled = enabled
        self._mode = mode  # REPORT | ENFORCE
        self._decision_cache: dict = {}
        self._http = httpx.AsyncClient(timeout=5.0)

    async def start_heartbeat(self, interval_seconds: int = 3600):
        """Send heartbeat every interval. Run as background task."""
        while True:
            await self.heartbeat()
            await asyncio.sleep(interval_seconds)

    async def heartbeat(self):
        if not self._enabled:
            return
        path = f"/api/v1/register/{self._app_id}"
        await self._http.get(f"{self._url}{path}", headers=self._sign("GET", path, b""))

    def _sign(self, method: str, path: str, body: bytes) -> dict:
        ts = datetime.now(timezone.utc).isoformat()
        body_hash = hashlib.sha256(body).hexdigest()
        message = f"{ts}\n{method}\n{path}\n{body_hash}"
        sig = hmac.new(self._secret.encode(), message.encode(), hashlib.sha256).hexdigest()
        return {
            "X-AIGP-Signature": f"hmac-sha256={sig}",
            "X-AIGP-Timestamp": ts,
            "X-AIGP-App-Id": self._app_id,
        }

    async def check(self, use_case: str, model_id: str, **kwargs) -> GovernanceDecision:
        """Pre-invocation policy check. Cached by use_case+model."""
        if not self._enabled:
            return GovernanceDecision({"decision": "ALLOW"})

        cache_key = f"{use_case}:{model_id}"
        if cached := self._decision_cache.get(cache_key):
            if time.time() - cached["ts"] < cached["ttl"]:
                return cached["decision"]

        path = "/api/v1/request"
        body = json.dumps({
            "protocol_version": "2.0", "message_type": "REQUEST",
            "app_id": self._app_id, "use_case": use_case,
            "model_id": model_id, **kwargs,
        }).encode()

        try:
            resp = await self._http.post(
                f"{self._url}{path}", content=body,
                headers={**self._sign("POST", path, body), "Content-Type": "application/json"},
            )
            decision = GovernanceDecision(resp.json())
            self._decision_cache[cache_key] = {"decision": decision, "ts": time.time(), "ttl": decision.ttl}
            return decision
        except Exception:
            # ENFORCE = fail-closed; REPORT = fail-open
            if self._mode == "ENFORCE":
                return GovernanceDecision({"decision": "DENY", "reason": "Governance authority unreachable (ENFORCE mode)"})
            return GovernanceDecision({"decision": "ALLOW", "reason": "Governance authority unreachable (REPORT mode)"})

    async def record(self, use_case: str, model_id: str,
                     input_tokens: int, output_tokens: int,
                     duration_ms: float, status: str = "SUCCESS", **kwargs) -> None:
        """Post-invocation activity record. Fire-and-forget."""
        if not self._enabled:
            return
        path = "/api/v1/record"
        body = json.dumps({
            "protocol_version": "2.0", "message_type": "RECORD",
            "app_id": self._app_id, "use_case": use_case,
            "model_id": model_id, "timestamp": datetime.now(timezone.utc).isoformat(),
            "duration_ms": int(duration_ms), "status": status,
            "tokens": {"input_tokens": input_tokens, "output_tokens": output_tokens,
                       "total_tokens": input_tokens + output_tokens},
        }).encode()
        try:
            await self._http.post(
                f"{self._url}{path}", content=body,
                headers={**self._sign("POST", path, body), "Content-Type": "application/json"},
            )
        except Exception:
            pass  # fire-and-forget

---

← 7. Registration Schema · Section index · 9. Governance Authority (governance-server) Schema →