"""TMGT integration examples: in-memory request handlers. Standard library only. Run: python3 mock_client.py This is not a production SDK. /examples/v1/ paths dispatch to in-memory functions. """ import copy import json import re class ExampleError(ValueError): pass def minor(value): if not isinstance(value, str) or not re.fullmatch(r"\d+(?:\.\d{1,2})?", value): raise ExampleError("Use a non-negative decimal string with at most two places.") whole, _, fraction = value.partition(".") return int(whole) * 100 + int(fraction.ljust(2, "0") or "0") def money(value): sign = "-" if value < 0 else "" return f"{sign}{abs(value) // 100}.{abs(value) % 100:02d}" def sample_id(value): if not isinstance(value, str) or not value.startswith("example_"): raise ExampleError("Only fictional example_* identifiers are accepted.") return value class MockTmgtClient: """An in-memory state machine, not an HTTP client.""" def __init__(self): self.wallets = {} self.intents = {} self.requests = {} self.events = [] self.policy = None self.sequence = {} def _id(self, kind): self.sequence[kind] = self.sequence.get(kind, 0) + 1 return f"example_{kind}_{self.sequence[kind]:03d}" def _event(self, intent_id, event_type): self.events.append({"id": self._id("event"), "intent_id": intent_id, "type": event_type, "sample_only": True}) def request(self, method, path, body=None): # Intentionally no URL handling, HTTP library, credential or transport. if not path.startswith("/examples/v1/"): raise ExampleError("Only local /examples/v1/ paths exist in this example.") result = self._dispatch(method.upper(), path, copy.deepcopy(body or {})) return copy.deepcopy(result) def _dispatch(self, method, path, body): if (method, path) == ("POST", "/examples/v1/auth/sessions"): return {"id": self._id("session"), "actor_id": sample_id(body["actor_id"]), "authenticated": False, "sample_only": True} if (method, path) == ("POST", "/examples/v1/wallets"): wallet_id = self._id("wallet") wallet = {"id": wallet_id, "owner_id": sample_id(body["owner_id"]), "network": sample_id(body["network"]), "address": self._id("address"), "asset": "example_asset", "balance": "100.00", "sample_only": True} self.wallets[wallet_id] = wallet return wallet match = re.fullmatch(r"/examples/v1/wallets/(example_[A-Za-z0-9_]+)", path) if method == "GET" and match: return self.wallets[match[1]] if (method, path) == ("POST", "/examples/v1/gas/budgets"): if body["currency"] != "USD": raise ExampleError("The example budget uses fictional USD accounting only.") return {"id": self._id("budget"), "project_id": sample_id(body["project_id"]), "currency": "USD", "limit": money(minor(body["limit"])), "spent": "0.00", "sample_only": True} if (method, path) == ("POST", "/examples/v1/policies"): version = self.policy["version"] + 1 if self.policy else 1 self.policy = {"id": self._id("policy"), "version": version, "max_amount": money(minor(body["max_amount"])), "asset": sample_id(body["asset"]), "sample_only": True} return self.policy if (method, path) == ("POST", "/examples/v1/intents"): request_id = sample_id(body["request_id"]) for key in ("wallet_id", "recipient", "asset"): sample_id(body[key]) if minor(body["amount"]) <= 0: raise ExampleError("Amount must be positive.") if body["wallet_id"] not in self.wallets: raise ExampleError("Unknown example wallet.") if request_id in self.requests: previous = self.intents[self.requests[request_id]] if previous["body"] != body: raise ExampleError("Idempotency conflict: the same request_id has different content.") return {"id": previous["id"], "status": previous["status"], "sample_only": True} intent_id = self._id("intent") self.intents[intent_id] = {"id": intent_id, "body": body, "status": "created"} self.requests[request_id] = intent_id return {"id": intent_id, "status": "created", "sample_only": True} match = re.fullmatch(r"/examples/v1/intents/(example_[A-Za-z0-9_]+)/(simulation|evaluation|execution)", path) if method == "POST" and match: intent = self.intents[match[1]] action = match[2] wallet = self.wallets[intent["body"]["wallet_id"]] amount = minor(intent["body"]["amount"]) if action == "execution" and "result" in intent: return intent["result"] if "result" in intent: raise ExampleError("The completed local intent cannot be re-evaluated.") if action == "simulation": intent["simulation"] = {"intent_id": intent["id"], "effect": { "balance_before": wallet["balance"], "balance_after": money(minor(wallet["balance"]) - amount), "asset": intent["body"]["asset"]}, "network_state_sample_only": True, "sample_only": True} intent.pop("evaluation", None) intent["status"] = "simulated" return intent["simulation"] if "simulation" not in intent: raise ExampleError("Run the local simulation first.") if not self.policy: raise ExampleError("No policy: denied by default.") if intent["simulation"]["effect"]["balance_before"] != wallet["balance"]: raise ExampleError("Wallet state changed; simulate and evaluate again.") if action == "evaluation": reasons = [] if intent["body"]["asset"] != self.policy["asset"] or intent["body"]["asset"] != wallet["asset"]: reasons.append("ASSET_NOT_ALLOWED") if amount > minor(self.policy["max_amount"]): reasons.append("AMOUNT_LIMIT") if amount > minor(wallet["balance"]): reasons.append("INSUFFICIENT_FICTIONAL_BALANCE") decision = "deny" if reasons else "allow" intent["evaluation"] = {"intent_id": intent["id"], "decision": decision, "policy_version": self.policy["version"], "reasons": reasons, "t1_status": "illustrative_not_scored", "sample_only": True} intent["status"] = "evaluated" self._event(intent["id"], "example.intent.evaluated") return intent["evaluation"] decision = intent.get("evaluation") if not decision or decision["decision"] != "allow": raise ExampleError("No positive local admission; execution blocked.") if decision["policy_version"] != self.policy["version"]: raise ExampleError("Policy changed; evaluate again.") wallet["balance"] = intent["simulation"]["effect"]["balance_after"] intent["status"] = "example_completed" intent["result"] = {"intent_id": intent["id"], "status": "example_completed", "cryptography_performed": False, "signature": None, "network_transaction": None, "sample_only": True} self._event(intent["id"], "example.intent.completed") return intent["result"] if (method, path) == ("GET", "/examples/v1/events"): return {"items": self.events, "sample_only": True} raise ExampleError("Unknown local example operation.") def run_example(): client = MockTmgtClient() client.request("POST", "/examples/v1/auth/sessions", {"actor_id": "example_user_001"}) wallet = client.request("POST", "/examples/v1/wallets", {"owner_id": "example_user_001", "network": "example_network"}) client.request("POST", "/examples/v1/gas/budgets", {"project_id": "example_project_001", "currency": "USD", "limit": "25.00"}) client.request("POST", "/examples/v1/policies", {"max_amount": "100.00", "asset": "example_asset"}) body = {"request_id": "example_request_001", "wallet_id": wallet["id"], "recipient": "example_recipient_001", "asset": "example_asset", "amount": "10.00"} intent = client.request("POST", "/examples/v1/intents", body) base = "/examples/v1/intents/" + intent["id"] client.request("POST", base + "/simulation") decision = client.request("POST", base + "/evaluation") result = client.request("POST", base + "/execution") replay = client.request("POST", base + "/execution") denied = client.request("POST", "/examples/v1/intents", {**body, "request_id": "example_request_002", "amount": "150.00"}) denied_base = "/examples/v1/intents/" + denied["id"] client.request("POST", denied_base + "/simulation") denial = client.request("POST", denied_base + "/evaluation") blocked = False try: client.request("POST", denied_base + "/execution") except ExampleError: blocked = True return {"mode": "local_example", "wallet_id": wallet["id"], "intent_id": intent["id"], "decision": decision["decision"], "status": result["status"], "network_requests": 0, "cryptographic_signatures": 0, "idempotent_replay": replay == result, "denied_decision": denial["decision"], "denied_execution_blocked": blocked, "events": len(client.request("GET", "/examples/v1/events")["items"])} if __name__ == "__main__": print(json.dumps(run_example(), ensure_ascii=False, indent=2))