"""*
* @file main.py
* @author nirholas/universal-crypto-mcp
* @copyright (c) 2026 nich
* @license MIT
* @repository universal-crypto-mcp
* @version 0.14.9.3
* @checksum n1ch-0las-4e49-4348-786274000000
"""
"""httpx e2e test client using x402 v2 SDK."""
import os
import json
import asyncio
from dotenv import load_dotenv
from eth_account import Account
# Import from new x402 package
from x402 import x402Client
from x402.http import decode_payment_response_header
from x402.http.clients import x402_httpx_transport
from x402.mechanisms.evm import EthAccountSigner
from x402.mechanisms.evm.exact import register_exact_evm_client
from x402.mechanisms.svm import KeypairSigner
from x402.mechanisms.svm.exact import register_exact_svm_client
import httpx
# Load environment variables
load_dotenv()
# Get environment variables
evm_private_key = os.getenv("EVM_PRIVATE_KEY")
svm_private_key = os.getenv("SVM_PRIVATE_KEY")
base_url = os.getenv("RESOURCE_SERVER_URL")
endpoint_path = os.getenv("ENDPOINT_PATH")
if not base_url or not endpoint_path:
error_result = {"success": False, "error": "Missing required environment variables"}
print(json.dumps(error_result))
exit(1)
if not evm_private_key and not svm_private_key:
error_result = {"success": False, "error": "At least one of EVM_PRIVATE_KEY or SVM_PRIVATE_KEY must be set"}
print(json.dumps(error_result))
exit(1)
async def main():
# Create x402 client
client = x402Client()
# Register EVM exact scheme if private key is available
if evm_private_key:
account = Account.from_key(evm_private_key)
evm_signer = EthAccountSigner(account)
register_exact_evm_client(client, evm_signer)
# Register SVM exact scheme if private key is available
if svm_private_key:
svm_signer = KeypairSigner.from_base58(svm_private_key)
register_exact_svm_client(client, svm_signer)
# Create httpx client with x402 payment transport and increased timeout
# Set timeout to 30 seconds to handle busy servers during test runs
timeout = httpx.Timeout(30.0, connect=10.0)
async with httpx.AsyncClient(
base_url=base_url,
timeout=timeout,
transport=x402_httpx_transport(client),
) as http_client:
# Make request
try:
response = await http_client.get(endpoint_path)
# Read the response content
content = response.content
response_data = json.loads(content.decode())
# Prepare result
result = {
"success": True,
"data": response_data,
"status_code": response.status_code,
"payment_response": None,
}
# Check for payment response header (V2: PAYMENT-RESPONSE, V1: X-PAYMENT-RESPONSE)
payment_header = response.headers.get("PAYMENT-RESPONSE") or response.headers.get("X-PAYMENT-RESPONSE")
if payment_header:
payment_response = decode_payment_response_header(payment_header)
result["payment_response"] = payment_response.model_dump()
# Output structured result as JSON for proxy to parse
print(json.dumps(result))
exit(0)
except Exception as e:
error_result = {
"success": False,
"error": str(e),
"status_code": getattr(e, "response", {}).get("status_code", None)
if hasattr(e, "response")
else None,
}
print(json.dumps(error_result))
exit(1)
if __name__ == "__main__":
asyncio.run(main())
""" EOF - nirholas/universal-crypto-mcp | 1493 """
