diff --git a/blockrun_llm/solana_client.py b/blockrun_llm/solana_client.py index 6757f3d..223cf44 100644 --- a/blockrun_llm/solana_client.py +++ b/blockrun_llm/solana_client.py @@ -44,15 +44,37 @@ XAuthorAnalyticsResponse, XCompareAuthorsResponse, ) -from .x402 import ( - create_solana_payment_payload, - extract_solana_payment_details, - parse_payment_required, -) from .solana_wallet import get_solana_public_key -from .validation import validate_api_url, sanitize_error_response, validate_resource_url +from .validation import validate_api_url, sanitize_error_response + +try: + from x402 import x402ClientSync + from x402.mechanisms.svm import KeypairSigner + from x402.mechanisms.svm.exact.register import register_exact_svm_client + from x402.http.utils import decode_payment_required_header, encode_payment_signature_header +except ImportError: + raise ImportError( + "Solana payment requires the x402 SDK. " + "Install with: pip install blockrun-llm[solana]" + ) SOLANA_API_URL = "https://sol.blockrun.ai/api" + + +def _create_signer(private_key: str) -> KeypairSigner: + """Create a KeypairSigner, handling both full keypair and seed-only formats.""" + try: + return KeypairSigner.from_base58(private_key) + except (ValueError, Exception): + # Fallback: key may be a raw seed (32 bytes) rather than a full keypair (64 bytes) + import base58 + from solders.keypair import Keypair + + secret = base58.b58decode(private_key) + keypair = Keypair.from_seed(secret[:32]) + return KeypairSigner(keypair) + + DEFAULT_MAX_TOKENS = 1024 DEFAULT_TIMEOUT = 60.0 @@ -95,6 +117,11 @@ def __init__( self._last_call_cost: float = 0.0 self._address: Optional[str] = None + # Initialize x402 SDK client for Solana payment signing + self._x402_client = x402ClientSync() + signer = _create_signer(self._private_key) + register_exact_svm_client(self._x402_client, signer, rpc_url=rpc_url) + def get_wallet_address(self) -> str: if not self._address: self._address = get_solana_public_key(self._private_key) @@ -165,6 +192,22 @@ def list_models(self) -> List[Dict[str, Any]]: resp.raise_for_status() return resp.json().get("data", []) + @staticmethod + def _extract_payment_header(response: httpx.Response) -> Optional[str]: + """Extract x402 payment header from a 402 response (header or body).""" + payment_header = response.headers.get("payment-required") + if not payment_header: + try: + import base64 + import json + + resp_body = response.json() + if resp_body.get("accepts") or resp_body.get("x402Version"): + payment_header = base64.b64encode(json.dumps(resp_body).encode()).decode() + except Exception: + pass + return payment_header + def _request_with_payment(self, endpoint: str, body: Dict[str, Any]) -> ChatResponse: url = f"{self._api_url}{endpoint}" headers = {"Content-Type": "application/json", "User-Agent": _get_user_agent()} @@ -196,56 +239,19 @@ def _request_with_payment(self, endpoint: str, body: Dict[str, Any]) -> ChatResp def _handle_payment_and_retry( self, url: str, body: Dict[str, Any], response: httpx.Response ) -> ChatResponse: - payment_header = response.headers.get("payment-required") - if not payment_header: - try: - import base64 - import json - - resp_body = response.json() - if resp_body.get("accepts") or resp_body.get("x402Version"): - payment_header = base64.b64encode(json.dumps(resp_body).encode()).decode() - except Exception: - pass - + payment_header = self._extract_payment_header(response) if not payment_header: raise PaymentError("402 response but no payment requirements found") - payment_required = parse_payment_required(payment_header) - details = extract_solana_payment_details(payment_required) - - if not details["network"].startswith("solana:"): - raise PaymentError( - f"Expected Solana network, got: {details['network']}. " - "Use LLMClient for Base payments." - ) - - fee_payer = (details.get("extra") or {}).get("feePayer") - if not fee_payer: - raise PaymentError("Missing feePayer in 402 extra field") - - resource_info = details.get("resource") or {} - resource_url = validate_resource_url( - resource_info.get("url") or f"{self._api_url}/v1/chat/completions", - self._api_url, - ) - - payment_payload = create_solana_payment_payload( - private_key=self._private_key, - recipient=details["recipient"], - amount=details["amount"], - fee_payer=fee_payer, - resource_url=resource_url, - resource_description=resource_info.get("description") or "BlockRun Solana AI API call", - max_timeout_seconds=details["max_timeout_seconds"], - extra=details.get("extra"), - rpc_url=self._rpc_url, - ) + # Use x402 SDK to decode 402 response and create signed payment + payment_required = decode_payment_required_header(payment_header) + payment_payload = self._x402_client.create_payment_payload(payment_required) + encoded_payment = encode_payment_signature_header(payment_payload) payment_headers = { "Content-Type": "application/json", "User-Agent": _get_user_agent(), - "PAYMENT-SIGNATURE": payment_payload, + "PAYMENT-SIGNATURE": encoded_payment, } # Retry with payment, with one automatic retry on 502/503 @@ -269,7 +275,7 @@ def _handle_payment_and_retry( sanitize_error_response(error_body), ) - cost_usd = float(details["amount"]) / 1e6 + cost_usd = float(payment_payload.accepted.amount) / 1e6 self._session_calls += 1 self._session_total_usd += cost_usd self._last_call_cost = cost_usd @@ -323,56 +329,19 @@ def _handle_payment_and_retry_raw( self, url: str, body: Dict[str, Any], response: httpx.Response ) -> Dict[str, Any]: """Handle 402 for raw endpoints with Solana payment.""" - payment_header = response.headers.get("payment-required") - if not payment_header: - try: - import base64 - import json - - resp_body = response.json() - if resp_body.get("accepts") or resp_body.get("x402Version"): - payment_header = base64.b64encode(json.dumps(resp_body).encode()).decode() - except Exception: - pass - + payment_header = self._extract_payment_header(response) if not payment_header: raise PaymentError("402 response but no payment requirements found") - payment_required = parse_payment_required(payment_header) - details = extract_solana_payment_details(payment_required) - - if not details["network"].startswith("solana:"): - raise PaymentError( - f"Expected Solana network, got: {details['network']}. " - "Use LLMClient for Base payments." - ) - - fee_payer = (details.get("extra") or {}).get("feePayer") - if not fee_payer: - raise PaymentError("Missing feePayer in 402 extra field") - - resource_info = details.get("resource") or {} - resource_url = validate_resource_url( - resource_info.get("url") or url, - self._api_url, - ) - - payment_payload = create_solana_payment_payload( - private_key=self._private_key, - recipient=details["recipient"], - amount=details["amount"], - fee_payer=fee_payer, - resource_url=resource_url, - resource_description=resource_info.get("description") or "BlockRun Solana AI API call", - max_timeout_seconds=details["max_timeout_seconds"], - extra=details.get("extra"), - rpc_url=self._rpc_url, - ) + # Use x402 SDK to decode 402 response and create signed payment + payment_required = decode_payment_required_header(payment_header) + payment_payload = self._x402_client.create_payment_payload(payment_required) + encoded_payment = encode_payment_signature_header(payment_payload) payment_headers = { "Content-Type": "application/json", "User-Agent": _get_user_agent(), - "PAYMENT-SIGNATURE": payment_payload, + "PAYMENT-SIGNATURE": encoded_payment, } # Retry with payment, with one automatic retry on 502/503 @@ -396,7 +365,7 @@ def _handle_payment_and_retry_raw( sanitize_error_response(error_body), ) - cost_usd = float(details["amount"]) / 1e6 + cost_usd = float(payment_payload.accepted.amount) / 1e6 self._session_calls += 1 self._session_total_usd += cost_usd self._last_call_cost = cost_usd diff --git a/blockrun_llm/x402.py b/blockrun_llm/x402.py index 647d545..36b32a8 100644 --- a/blockrun_llm/x402.py +++ b/blockrun_llm/x402.py @@ -236,212 +236,13 @@ def extract_payment_details(payment_required: Dict[str, Any]) -> Dict[str, Any]: # ============================================================ -# Solana x402 Payment +# Solana x402 Payment — delegated to official x402 SDK # ============================================================ - -SOLANA_NETWORK = "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp" -USDC_SOLANA = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" - -# SPL program IDs -TOKEN_PROGRAM_ID = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" -ASSOCIATED_TOKEN_PROGRAM_ID = "ATokenGPvbdGVxr1b2hvZbsiqW5xWH25efTNsLJe1bRS" - -# Compute budget defaults (match @x402/svm) -DEFAULT_COMPUTE_UNIT_PRICE_MICROLAMPORTS = 1 -DEFAULT_COMPUTE_UNIT_LIMIT = 8000 - - -def _get_ata(owner: str, mint: str) -> str: - """Derive Associated Token Account address.""" - from solders.pubkey import Pubkey # type: ignore - - owner_pk = Pubkey.from_string(owner) - mint_pk = Pubkey.from_string(mint) - token_program = Pubkey.from_string(TOKEN_PROGRAM_ID) - assoc_program = Pubkey.from_string(ASSOCIATED_TOKEN_PROGRAM_ID) - - seeds = [bytes(owner_pk), bytes(token_program), bytes(mint_pk)] - ata, _ = Pubkey.find_program_address(seeds, assoc_program) - return str(ata) - - -def _get_latest_blockhash(rpc_url: str) -> str: - """Fetch latest blockhash from Solana RPC.""" - import httpx - - resp = httpx.post( - rpc_url, - json={ - "jsonrpc": "2.0", - "id": 1, - "method": "getLatestBlockhash", - "params": [{"commitment": "finalized"}], - }, - timeout=10, - ) - resp.raise_for_status() - return resp.json()["result"]["value"]["blockhash"] - - -def create_solana_payment_payload( - private_key: str, - recipient: str, - amount: str, - fee_payer: str, - resource_url: str = "https://sol.blockrun.ai/api/v1/chat/completions", - resource_description: str = "BlockRun Solana AI API call", - max_timeout_seconds: int = 300, - extra: Optional[Dict[str, Any]] = None, - extensions: Optional[Dict[str, Any]] = None, - rpc_url: str = "https://api.mainnet-beta.solana.com", -) -> str: - """ - Create a signed Solana x402 v2 payment payload. - - Builds an SPL TransferChecked transaction signed by the user's Solana keypair. - The CDP facilitator (feePayer) co-signs on the server side. - - Args: - private_key: bs58-encoded 64-byte Solana secret key - recipient: Payment recipient Solana address (base58) - amount: Amount in micro USDC (6 decimals, e.g. "1000" = $0.001) - fee_payer: CDP facilitator address that pays SOL transaction fees (base58) - resource_url: URL of the resource being accessed - resource_description: Description for the payment - max_timeout_seconds: Max timeout for the payment - extra: Extra info included in payment (e.g. feePayer) - extensions: x402 extensions dict - rpc_url: Solana RPC endpoint - - Returns: - Base64-encoded signed payment payload - """ - try: - from solders.keypair import Keypair # type: ignore - from solders.pubkey import Pubkey # type: ignore - from solders.hash import Hash # type: ignore - from solders.instruction import Instruction, AccountMeta # type: ignore - from solders.message import MessageV0 # type: ignore - from solders.transaction import VersionedTransaction # type: ignore - from solders.signature import Signature # type: ignore - import base58 # type: ignore - except ImportError: - raise ImportError( - "Solana payment requires 'solders' and 'base58'. " - "Install with: pip install blockrun-llm[solana]" - ) - - # Load keypair from first 32 bytes (seed) - secret = base58.b58decode(private_key) - keypair = Keypair.from_seed(secret[:32]) - owner_pubkey = keypair.pubkey() - - # Derive ATAs - source_ata = _get_ata(str(owner_pubkey), USDC_SOLANA) - dest_ata = _get_ata(recipient, USDC_SOLANA) - - # Get latest blockhash - blockhash = _get_latest_blockhash(rpc_url) - - # Build compute budget instructions - compute_budget_id = Pubkey.from_string("ComputeBudget111111111111111111111111111111") - - # setComputeUnitLimit instruction: discriminator=2, units=u32 LE - import struct - - limit_data = bytes([2]) + struct.pack(" bool: """Check if a network string represents Solana.""" return network.startswith("solana:") - - -def extract_solana_payment_details(payment_required: Dict[str, Any]) -> Dict[str, Any]: - """ - Extract Solana payment details from a 402 response. - Finds the Solana network option in accepts[]. - """ - accepts = payment_required.get("accepts", []) - option = next((o for o in accepts if is_solana_network(o.get("network", ""))), None) - if not option: - raise ValueError("No Solana payment option found in 402 response") - - amount = option.get("amount") or option.get("maxAmountRequired") - if not amount: - raise ValueError("No amount in Solana payment requirements") - - return { - "amount": amount, - "recipient": option.get("payTo"), - "network": option.get("network"), - "asset": option.get("asset"), - "max_timeout_seconds": option.get("maxTimeoutSeconds", 300), - "extra": option.get("extra", {}), - "resource": payment_required.get("resource"), - } diff --git a/pyproject.toml b/pyproject.toml index 99779ed..4e475f0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -41,8 +41,7 @@ dev = [ "ruff>=0.1.0", ] solana = [ - "solders>=0.21.0", - "base58>=2.1.0", + "x402[svm]>=2.0.0", ] [project.urls] diff --git a/tests/unit/test_x402.py b/tests/unit/test_x402.py index ec6a846..62ca893 100644 --- a/tests/unit/test_x402.py +++ b/tests/unit/test_x402.py @@ -223,48 +223,70 @@ def test_include_resource(self): assert details["resource"]["url"] == "https://api.blockrun.ai/test" -class TestCreateSolanaPaymentPayload: - """Tests for Solana payment payload creation.""" +class TestSolanaX402SdkIntegration: + """Tests for Solana x402 SDK integration.""" - TEST_BS58_KEY = ( - "5MaiiCavjCmn9Hs1o3eznqDEhRwxo7pXiAYez7keQUviQeRjpzKCY8trDwpvBMTKTpNFbCJsBZthJ4tCs6o62rr" - ) + USDC_SOLANA = "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" + TOKEN_PROGRAM_ID = "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA" TEST_FEE_PAYER = "2wKupLR9q6wXYppw8Gr2NvWxKBUqm4PPJKkQfoxHDBg4" - TEST_RECIPIENT = "AQqnMFBwGZEoti85aTVRy8XYpKrho7GaMDx9ZB3CEeKA" - - def test_payload_structure(self): - """Should create valid Solana payment payload.""" - from blockrun_llm.x402 import create_solana_payment_payload - import json - import base64 - - payload = create_solana_payment_payload( - private_key=self.TEST_BS58_KEY, - recipient=self.TEST_RECIPIENT, - amount="1000", - fee_payer=self.TEST_FEE_PAYER, - ) + TEST_SOL_RECIPIENT = "AQqnMFBwGZEoti85aTVRy8XYpKrho7GaMDx9ZB3CEeKA" - assert isinstance(payload, str) - decoded = json.loads(base64.b64decode(payload)) - assert decoded["x402Version"] == 2 - assert "transaction" in decoded["payload"] - assert decoded["accepted"]["network"].startswith("solana:") - assert decoded["accepted"]["asset"] == "EPjFWdd5AufqSSqeM2qN1xzybapC8G4wEGGkZwyTDt1v" - - def test_payload_transaction_is_base64(self): - """Transaction field should be base64-encoded.""" - from blockrun_llm.x402 import create_solana_payment_payload - import json - import base64 - - payload = create_solana_payment_payload( - private_key=self.TEST_BS58_KEY, - recipient=self.TEST_RECIPIENT, - amount="1000", - fee_payer=self.TEST_FEE_PAYER, - ) - decoded = json.loads(base64.b64decode(payload)) - # Should be valid base64 - tx_bytes = base64.b64decode(decoded["payload"]["transaction"]) - assert len(tx_bytes) > 0 + def test_decode_solana_payment_required(self): + """Should decode a Solana 402 PaymentRequired header.""" + from x402.http.utils import decode_payment_required_header + + data = { + "x402Version": 2, + "accepts": [ + { + "scheme": "exact", + "network": "solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp", + "amount": "1000", + "asset": self.USDC_SOLANA, + "payTo": self.TEST_SOL_RECIPIENT, + "maxTimeoutSeconds": 300, + "extra": {"feePayer": self.TEST_FEE_PAYER}, + } + ], + } + encoded = base64.b64encode(json.dumps(data).encode()).decode() + result = decode_payment_required_header(encoded) + + assert result.x402_version == 2 + assert len(result.accepts) == 1 + assert str(result.accepts[0].network).startswith("solana:") + assert result.accepts[0].pay_to == self.TEST_SOL_RECIPIENT + assert result.accepts[0].amount == "1000" + assert result.accepts[0].extra["feePayer"] == self.TEST_FEE_PAYER + + def test_keypair_signer_address(self): + """KeypairSigner should derive correct public key from bs58 secret.""" + from x402.mechanisms.svm import KeypairSigner + from solders.keypair import Keypair + + # Generate a valid keypair and get its base58 representation + kp = Keypair() + expected_address = str(kp.pubkey()) + + signer = KeypairSigner.from_base58(str(kp)) + assert signer.address == expected_address + + def test_ata_derivation_uses_correct_program_id(self): + """ATA derivation must use the correct Associated Token Program ID.""" + from x402.mechanisms.svm import derive_ata + + # Known wallet -> known USDC ATA (verified on-chain) + owner = "CtJTYWPQSL5jw9B2JRHmpQjYCSSgUX3LRvmMBhq55HmQ" + expected_ata = "HZPPxg9ZyoHu4f2pj5uEEXsArLA2rnL9FtDgC8rrAp5Q" + + result = derive_ata(owner, self.USDC_SOLANA, self.TOKEN_PROGRAM_ID) + assert result == expected_ata + + def test_is_solana_network(self): + """Should correctly identify Solana networks.""" + from blockrun_llm.x402 import is_solana_network + + assert is_solana_network("solana:5eykt4UsFv8P8NJdTREpY1vzqKqZKvdp") + assert is_solana_network("solana:EtWTRABZaYq6iMfeYKouRu166VU2xqa1") + assert not is_solana_network("eip155:8453") + assert not is_solana_network("base-sepolia")