Upload builderbrain/circle_gateway_client.py

builderbrain/circle_gateway_client.py

@@ -0,0 +1,228 @@
+"""
+Real Circle Gateway API Client
+==============================
+
+Implements the actual Circle Gateway HTTP API for:
+- Cross-chain USDC transfers (CCTP via Gateway)
+- Unified balance queries
+- Domain enumeration
+
+API Endpoints (from docs):
+- GET  https://gateway-api-testnet.circle.com/v1/gateway-info
+- POST https://gateway-api-testnet.circle.com/v1/balances
+- POST https://gateway-api-testnet.circle.com/v1/transfer
+
+Auth: Bearer <CIRCLE_API_KEY>
+
+References:
+- Circle Gateway Docs: https://developers.circle.com/gateway
+- Nanopayments Blog: https://community.arc.network/public/clubs/agentic-economy/blogs/...
+"""
+
+import requests
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+from datetime import datetime
+
+
+@dataclass
+class GatewayDomain:
+    """A supported domain (chain) from Gateway info."""
+    chain: str           # e.g. "Ethereum"
+    network: str         # e.g. "Sepolia"
+    domain: int          # uint32 domain ID
+    wallet_contract: str
+    minter_contract: str
+    supported_tokens: List[str]
+
+
+@dataclass
+class GatewayBalance:
+    """USDC balance in a Gateway wallet contract."""
+    domain: int
+    depositor: str
+    balance: str  # decimal string, e.g. "123.456789"
+
+
+@dataclass
+class TransferAttestation:
+    """Response from POST /v1/transfer (create transfer attestation)."""
+    transfer_id: str
+    attestation: str          # hex-encoded bytes
+    signature: str            # operator ECDSA signature
+    total_fees: str           # decimal string in USDC
+    fee_token: str            # "USDC"
+    per_intent_fees: List[Dict]
+    expiration_block: int
+
+
+class CircleGatewayClient:
+    """
+    Production client for Circle Gateway API (testnet).
+    
+    Usage:
+        client = CircleGatewayClient(api_key="sk_test_...")
+        domains = client.get_gateway_info()
+        balances = client.get_balances(depositor="0x...", domain=0)
+        attestation = client.create_transfer(...)
+    """
+    
+    TESTNET_BASE = "https://gateway-api-testnet.circle.com"
+    
+    def __init__(self, api_key: str, base_url: Optional[str] = None):
+        self.api_key = api_key
+        self.base_url = base_url or self.TESTNET_BASE
+        self.session = requests.Session()
+        self.session.headers.update({
+            "Authorization": f"Bearer {api_key}",
+            "Content-Type": "application/json",
+        })
+    
+    # ────────────────────────────── Gateway Info ──────────────────────────────
+    
+    def get_gateway_info(self) -> List[GatewayDomain]:
+        """
+        GET /v1/gateway-info
+        
+        Returns supported domains with contract addresses.
+        Use this instead of hardcoding chains.
+        """
+        resp = self.session.get(
+            f"{self.base_url}/v1/gateway-info",
+            timeout=15,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        
+        domains = []
+        for d in data.get("domains", []):
+            wc = d.get("walletContract", {})
+            mc = d.get("minterContract", {})
+            domains.append(GatewayDomain(
+                chain=d.get("chain", ""),
+                network=d.get("network", ""),
+                domain=d.get("domain", 0),
+                wallet_contract=wc.get("address", ""),
+                minter_contract=mc.get("address", ""),
+                supported_tokens=wc.get("supportedTokens", []),
+            ))
+        return domains
+    
+    # ────────────────────────────── Balances ──────────────────────────────
+    
+    def get_balances(
+        self,
+        depositor: str,
+        domain: int,
+        token: str = "USDC",
+    ) -> List[GatewayBalance]:
+        """
+        POST /v1/balances
+        
+        Query unified USDC balance for a depositor on a domain.
+        
+        Request:
+            {
+              "token": "USDC",
+              "sources": [
+                {"depositor": "0x...", "domain": 0}
+              ]
+            }
+        """
+        payload = {
+            "token": token,
+            "sources": [
+                {"depositor": depositor, "domain": domain}
+            ],
+        }
+        
+        resp = self.session.post(
+            f"{self.base_url}/v1/balances",
+            json=payload,
+            timeout=15,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        
+        balances = []
+        for b in data.get("balances", []):
+            balances.append(GatewayBalance(
+                domain=b.get("domain", 0),
+                depositor=b.get("depositor", ""),
+                balance=b.get("balance", "0"),
+            ))
+        return balances
+    
+    # ────────────────────────────── Transfer Attestation ──────────────────────────────
+    
+    def create_transfer_attestation(
+        self,
+        burn_intents: List[Dict],
+    ) -> TransferAttestation:
+        """
+        POST /v1/transfer
+        
+        Create a transfer attestation for one or more burn intents.
+        
+        WARNING: This is the LOW-LEVEL protocol API. Each burn intent requires:
+        - Source wallet domain + wallet contract address
+        - Destination minter domain + minter contract address  
+        - Token addresses (source and dest)
+        - Debitor (depositor) and recipient addresses (32-byte padded)
+        - Amount, max fee, max block height, user salt
+        - ECDSA signature over encoded burn intent(s)
+        
+        For high-level transfers, use Circle's App Kit or Bridge SDK instead.
+        
+        This method is a PLACEHOLDER for the full encoded intent construction.
+        In production, you would:
+        1. Fetch domain info from get_gateway_info()
+        2. Encode burn intent per CCTP spec
+        3. Sign with depositor's key (or Circle Wallet API)
+        4. Submit encoded intent + signature
+        """
+        # TODO: Implement full burn intent encoding + signing
+        # For hackathon, this is intentionally left as a documented stub
+        # because encoding requires domain-specific contract addresses
+        # and depositor signatures that depend on your wallet setup.
+        
+        resp = self.session.post(
+            f"{self.base_url}/v1/transfer",
+            json={"burnIntents": burn_intents},
+            timeout=30,
+        )
+        resp.raise_for_status()
+        data = resp.json()
+        
+        return TransferAttestation(
+            transfer_id=data.get("transferId", ""),
+            attestation=data.get("attestation", ""),
+            signature=data.get("signature", ""),
+            total_fees=data.get("fees", {}).get("total", "0"),
+            fee_token=data.get("fees", {}).get("token", "USDC"),
+            per_intent_fees=data.get("fees", {}).get("perIntent", []),
+            expiration_block=data.get("expirationBlock", 0),
+        )
+    
+    def mint_on_destination(
+        self,
+        minter_contract_address: str,
+        attestation: str,
+        signature: str,
+    ) -> Dict:
+        """
+        On-chain call: gatewayMinter.gatewayMint(attestation, signature)
+        
+        This is an on-chain contract call, NOT a REST API call.
+        Use your Web3/Ethers client to call the minter contract on the
+        destination chain.
+        
+        Returns transaction hash.
+        """
+        # This is a documentation-only method; actual implementation
+        # requires Web3 provider + contract ABI + depositor wallet
+        raise NotImplementedError(
+            "On-chain minting requires Web3 provider. "
+            "Call gatewayMinter.gatewayMint(attestation, signature) "
+            "on the destination chain's minter contract."
+        )