g402_tool

agentipy/g402/.env.example

@@ -0,0 +1,3 @@
+SERVER_WALLET_ADDRESS=""
+
+CLIENT_WALLET_PRIVATE_KEY_BASE58=""

agentipy/g402/README.md

@@ -0,0 +1,83 @@
+# G402 Solana Micropayment Protocol with Agentipy
+
+This project is a complete Python implementation of the HTTP 402 Payment Required protocol using a direct on-chain Solana micropayment flow.
+
+It has been fully integrated with the `agentipy` framework, providing a reusable `G402DataFetcher` tool that allows autonomous Solana agents to interact with payment-gated APIs.
+
+## Key Features
+
+-   **Agentipy Integration**: The client logic is packaged as a native `agentipy` tool, making it easy for agents to purchase access to protected data.
+-   **Clean SDK Architecture**: The server-side logic is contained in a `SolanaPaymentGuard`, making it simple to protect any FastAPI endpoint.
+-   **Modern Solana Transactions**: Uses Versioned Transactions and priority fees to ensure fast and reliable transaction processing.
+-   **Robust Confirmation**: The agent's tool waits for `Finalized` commitment, preventing timeouts and ensuring payment certainty.
+-   **Replay Attack Protection**: The server validates and stores unique payment references to prevent a single payment from being used multiple times.
+
+## How It Works
+
+1.  The `run_agent.py` script initializes a `SolanaAgentKit` and instructs it to use the `G402DataFetcher` tool to access a protected URL.
+2.  The `server.py`, protected by the `SolanaPaymentGuard`, receives the request and returns a **`402 Payment Required`** response containing payment details (receiver, amount, reference).
+3.  The `G402DataFetcher` tool handles the 402 response, uses the agent's wallet to construct and send a Solana transaction, and waits for it to be finalized.
+4.  Once confirmed, the tool automatically retries the request with the transaction signature and reference in the headers as proof of payment.
+5.  The `SolanaPaymentGuard` verifies the transaction on-chain. If valid, it allows the request to proceed.
+6.  The server returns a **`200 OK`** response with the premium data, which is then returned to the agent.
+
+## Setup Guide
+
+### 1. Clone & Navigate
+
+```bash
+git clone https://github.com/agentipy/agentipy/g402
+cd g402
+```
+
+### 2. Create Virtual Environment
+
+```bash
+python -m venv .venv
+
+# On macOS/Linux:
+source .venv/bin/activate
+
+# On Windows PowerShell:
+.venv\Scripts\Activate.ps1
+```
+
+### 3. Install Dependencies
+
+```bash
+pip install -r requirements.txt
+```
+
+### 4. Configure Environment
+
+Create a `.env` file and copy the contents of `.env.example` into it. Fill in your server's public key and your client's base58 private key.
+
+### 5. Add the Custom Tool to Agentipy
+
+You must place the `use_g402.py` tool inside the installed `agentipy` library.
+
+First, find the library's location:
+```bash
+pip show agentipy
+```
+
+
+## Running the Application
+
+You must run the server and the agent in two separate terminals. Ensure the virtual environment is activated in both.
+
+### Terminal 1: Start the Server
+
+```bash
+uvicorn server:app --reload
+```
+
+### Terminal 2: Run the Agent
+
+Once the server is running, execute the agent script:
+
+```bash
+python run_agent.py
+```
+
+The agent will proceed through the entire payment flow, printing its status at each step and displaying the final unlocked data.

agentipy/g402/client.py

@@ -0,0 +1,22 @@
+import os
+from dotenv import load_dotenv
+from solders.keypair import Keypair
+from g402_cl.buyer_sdk import SolanaPaymentClient
+
+load_dotenv()
+SERVER_URL = "http://127.0.0.1:8000/premium-data"
+RPC_URL = "https://api.devnet.solana.com"
+
+secret_b58 = os.getenv("CLIENT_WALLET_PRIVATE_KEY_BASE58")
+if not secret_b58:
+    raise SystemExit("Error: Set CLIENT_WALLET_PRIVATE_KEY_BASE58 in your .env file")
+
+client_keypair = Keypair.from_base58_string(secret_b58)
+
+if __name__ == "__main__":
+    payment_client = SolanaPaymentClient(
+        rpc_url=RPC_URL,
+        client_keypair=client_keypair
+    )
+
+    payment_client.get_premium_data(server_url=SERVER_URL)

agentipy/g402/g402_cl/buyer_sdk.py

@@ -0,0 +1,136 @@
+# g402_cl/buyer_sdk.py
+
+import time
+import requests
+from solders.keypair import Keypair
+from solders.pubkey import Pubkey
+from solders.signature import Signature
+from solders.system_program import TransferParams, transfer
+from solders.transaction import VersionedTransaction
+from solders.message import MessageV0
+from solders.compute_budget import set_compute_unit_price
+from solana.rpc.api import Client
+from solana.rpc.types import TxOpts
+from solana.rpc.commitment import Confirmed, Finalized # Import Finalized
+
+class SolanaPaymentClient:
+    def __init__(self, rpc_url: str, client_keypair: Keypair):
+        self.client = Client(rpc_url, timeout=30.0)
+        self.keypair = client_keypair
+        print("✅ Buyer SDK: SolanaPaymentClient initialized.")
+        print(f"   - Pubkey: {self.keypair.pubkey()}")
+        try:
+            bal = self.client.get_balance(self.keypair.pubkey()).value
+            print(f"   - Balance: {bal / 1_000_000_000:.9f} SOL")
+        except Exception as e:
+            print(f"⚠️  Could not fetch wallet balance. Check RPC connection. Error: {e}")
+
+    def _wait_for_confirmation(self, sig: str, timeout: int = 60):
+        """Robust confirmation waiting using client.confirm_transaction"""
+        print(f"Waiting for confirmation of tx: {sig} (timeout: {timeout}s)...")
+        try:
+            signature_to_check = Signature.from_string(sig)
+            # Use the library's built-in confirmation function
+            confirmation_resp = self.client.confirm_transaction(
+                signature_to_check,
+                commitment=Finalized,  # Wait for finalization for higher certainty
+                sleep_seconds=2,  # How long to wait between checks
+                last_valid_block_height=self.client.get_latest_blockhash().value.last_valid_block_height
+            )
+            if confirmation_resp.value[0].err:
+                raise SystemExit(f"❌ Transaction failed: {confirmation_resp.value[0].err}")
+
+            print("✅ Transaction Confirmed!")
+            return True
+
+        except Exception as e:
+            explorer_url = f"https://explorer.solana.com/tx/{sig}?cluster=devnet"
+            print(f"An error occurred during confirmation: {e}")
+            raise SystemExit(f"❌ Tx confirmation failed or timed out: {explorer_url}")
+
+
+    def _send_payment(self, to_addr: str, lamports: int) -> str:
+        print(f"   → Paying {lamports / 1_000_000_000:.6f} SOL → {to_addr[:8]}...")
+
+        # 1. FRESH blockhash (critical!)
+        blockhash_resp = self.client.get_latest_blockhash(commitment=Confirmed)
+        recent_blockhash = blockhash_resp.value.blockhash
+
+        # 2. Priority fee: 200k micro-lamports = lands in <5s on Devnet
+        priority_ix = set_compute_unit_price(200_000)
+
+        # 3. Transfer
+        transfer_ix = transfer(TransferParams(
+            from_pubkey=self.keypair.pubkey(),
+            to_pubkey=Pubkey.from_string(to_addr),
+            lamports=lamports
+        ))
+
+        # 4. Build message
+        msg = MessageV0.try_compile(
+            payer=self.keypair.pubkey(),
+            instructions=[priority_ix, transfer_ix],
+            address_lookup_table_accounts=[],
+            recent_blockhash=recent_blockhash,
+        )
+
+        tx = VersionedTransaction(msg, [self.keypair])
+        raw_tx = bytes(tx)
+
+        # 5. Send with preflight + retry
+        opts = TxOpts(skip_preflight=False, preflight_commitment=Confirmed)
+        sig = self.client.send_raw_transaction(raw_tx, opts).value
+
+        print(f"   Payment sent! Signature: {sig}")
+        self._wait_for_confirmation(str(sig))
+        return str(sig)
+
+    def get_premium_data(self, server_url: str):
+        """Main method: Get 402 → Pay → Retry with proof"""
+        session = requests.Session()
+        print(f"\n{'='*60}")
+        print(f"STEP 1: Requesting data from {server_url} (expecting 402)...")
+        print(f"{'='*60}")
+
+        try:
+            initial_response = session.get(server_url, timeout=10)
+        except requests.exceptions.RequestException as e:
+            raise SystemExit(f"❌ Connection error: {e}")
+
+        if initial_response.status_code != 402:
+            raise SystemExit(
+                f"❌ Unexpected status: {initial_response.status_code}\n"
+                f"{initial_response.text}"
+            )
+
+        details = initial_response.json()
+        print("✅ Received 402. Payment Details:", details)
+
+        print(f"\n{'='*60}")
+        print("STEP 2: Sending payment on Solana devnet...")
+        print(f"{'='*60}")
+        tx_sig = self._send_payment(
+            details["receiver"],
+            details["amount_lamports"]
+        )
+
+        print(f"\n{'='*60}")
+        print("STEP 3: Retrying request with payment proof...")
+        print(f"{'='*60}")
+
+        session.headers.update({
+            "X-Payment-Signature": tx_sig,
+            "X-Payment-Reference": details["reference"],
+        })
+
+        final_response = session.get(server_url, timeout=10)
+        print(f"\n🎉 FINAL RESPONSE ({final_response.status_code}):")
+        print(final_response.json())
+        print(f"{'='*60}")
+
+        if final_response.status_code == 200:
+            print("🚀 SUCCESS! Payment accepted, premium data unlocked!")
+        else:
+            print("⚠️  Server rejected payment proof. Check server logs.")
+
+        return final_response.json()

agentipy/g402/g402_cl/seller_sdk.py

@@ -0,0 +1,86 @@
+import time
+import uuid
+import json
+from fastapi import Request, status, Response
+from fastapi.responses import JSONResponse
+from solders.signature import Signature
+from solana.rpc.api import Client
+from solana.rpc.commitment import Confirmed
+from solana.exceptions import SolanaRpcException
+
+processed_references = set()
+
+class SolanaPaymentGuard:
+    def __init__(self, rpc_url: str, server_wallet: str, price_lamports: int):
+        self.client = Client(rpc_url)
+        self.server_wallet = server_wallet
+        self.price_lamports = price_lamports
+        print("Seller SDK: SolanaPaymentGuard initialized.")
+
+    def _find_system_transfer(self, tx_info: dict):
+        try:
+            transaction_data = tx_info.get('transaction', {})
+            msg = transaction_data.get('message', {})
+            instructions = msg.get('instructions', [])
+            for ix in instructions:
+                if ix.get("program") == "system" and ix.get("parsed", {}).get("type") == "transfer":
+                    info = ix.get("parsed", {}).get("info", {})
+                    destination = info.get("destination")
+                    lamports = info.get("lamports")
+                    if destination and lamports is not None:
+                        return destination, lamports
+        except Exception as e:
+            print(f"Error parsing transaction: {e}")
+        return None, 0
+
+    async def require_payment(self, request: Request):
+        sig_header = request.headers.get("X-Payment-Signature")
+        ref_header = request.headers.get("X-Payment-Reference")
+
+        if not sig_header or not ref_header:
+            reference = str(uuid.uuid4())
+            return JSONResponse(
+                status_code=status.HTTP_402_PAYMENT_REQUIRED,
+                content={
+                    "message": "Payment Required",
+                    "receiver": self.server_wallet,
+                    "amount_lamports": self.price_lamports,
+                    "reference": reference,
+                },
+            )
+
+        if ref_header in processed_references:
+            raise ValueError("This payment reference has already been used.")
+
+        sig = Signature.from_string(sig_header.strip())
+
+        tx_info = None
+        for attempt in range(5):
+            try:
+                resp = self.client.get_transaction(
+                    sig, "jsonParsed", max_supported_transaction_version=0, commitment=Confirmed
+                )
+                if resp.value:
+                    tx_info = resp.value
+                    break
+            except SolanaRpcException as e:
+                print(f"RPC error on attempt {attempt+1}: {e}")
+            time.sleep(2)
+
+        if not tx_info:
+            raise ValueError("Transaction not found or not confirmed.")
+
+        if tx_info.transaction.meta and tx_info.transaction.meta.err:
+            raise ValueError(f"Transaction failed on-chain: {tx_info.transaction.meta.err}")
+
+        tx_info_dict = json.loads(tx_info.to_json())
+        to_addr, lamports = self._find_system_transfer(tx_info_dict)
+
+        if to_addr != self.server_wallet:
+            raise ValueError(f"Payment sent to wrong receiver. Expected {self.server_wallet}, got {to_addr}.")
+        if lamports < self.price_lamports:
+            raise ValueError(f"Insufficient amount paid. Expected {self.price_lamports}, got {lamports}.")
+
+        processed_references.add(ref_header)
+        print(f"SDK: Payment verified for {lamports} lamports. Signature: {sig}")
+        return {"signature": str(sig), "amount": lamports}

agentipy/g402/requirements.txt

@@ -0,0 +1,9 @@
+fastapi
+uvicorn[standard]
+python-dotenv
+solana
+solders
+requests
+agentipy
+httpx
+