Solana Jupiter API兑换报错：未找到swapTransaction的解决方案

问题描述 #

编写基于Solana和Jupiter API的SOL兑换代币代码时，运行出现「No swapTransaction found in response」错误。返回的quote数据包含路由规划等信息，但缺失swapTransaction字段，无法完成兑换流程。

原代码 #

import requests
import base64
from base58 import b58decode
from solana.rpc.api import Client
from solana.transaction import Transaction
from solders.keypair import Keypair
import json

# Solana Configuration
solana_client = Client("https://mainnet.helius-rpc.com/?api-key=API")  # You may use a different endpoint if needed
private_key_base58 = "KEY"  # Private Key
private_key_bytes = b58decode(private_key_base58)  # Base58

# Jupiter API Configuration
JUPITER_API_URL = "https://quote-api.jup.ag/v6/quote"
SWAP_API_URL = "https://quote-api.jup.ag/v6/swap"

# Slippage
slippage = 0.1  # 10%

# Private Key 64 bytes
if len(private_key_bytes) == 64:
    payer = Keypair.from_bytes(private_key_bytes)
    print(f"Public Key: {payer.pubkey()}")
else:
    print("Error: Private key length is invalid. It should be 64 bytes.")

amount_to_swap = 0.01  # SOL

# ========== FUNCTIONALITY ==========

# Contract Address
def extract_contract_address(message):
    pattern = r'\b[1-9A-HJ-NP-Za-km-z]{32,44}\b'
    match = re.search(pattern, message)
    if match:
        return match.group(0)
    return None

# quote from Jupiter API
def get_quote(input_mint, output_mint, amount, slippage):
    params = {
        'inputMint': input_mint,
        'outputMint': output_mint,
        'amount': str(int(amount * 10**9)),  # Convert to lamports
        'slippageBps': str(int(slippage * 10000))  # slippage in bps
    }

    response = requests.get(JUPITER_API_URL, params=params)
    if response.status_code == 200:
        data = response.json()
        print("Quote Data:", json.dumps(data, indent=2))  # ดูข้อมูลตอบกลับทั้งหมด
        if 'swapTransaction' in data:
            print("Swap Transaction:", data['swapTransaction'])
        else:
            print("No swapTransaction found in response.")
        return data
    else:
        print(f"Error fetching quote: {response.status_code}")
        return None

#  swap to Jupiter API
def swap_sol_to_token(input_amount, input_token_mint, output_token_mint):
    print(f"Swapping {input_amount} SOL for Token at Contract Address: {output_token_mint}...")

    try:
        #  Quote from Jupiter API
        quote_data = get_quote(input_token_mint, output_token_mint, input_amount, slippage)
        
        if not quote_data:
            print("Error: No valid data received from the quote API.")
            return

        # Check swapTransaction
        if 'swapTransaction' not in quote_data or not quote_data['swapTransaction']:
            print("Error: No swapTransaction found in response.")
            print("Quote Response Data:", json.dumps(quote_data, indent=2))
            return

        swap_transaction = quote_data['swapTransaction']
        print(f"Swap transaction data: {swap_transaction}")

        # Step 1: swap to Jupiter API
        swap_payload = {
            'quoteResponse': quote_data,
            'userPublicKey': str(payer.pubkey())
        }

        swap_response = requests.post(SWAP_API_URL, json=swap_payload)
        swap_response_data = swap_response.json()

        if swap_response.status_code == 200 and swap_response_data.get('swapTransaction'):
            swap_transaction = swap_response_data['swapTransaction']
            print(f"Swap Transaction: {swap_transaction}")

            # Step 2: Sendtx to Solana Blockchain
            raw_transaction = base64.b64decode(swap_transaction)
            transaction = Transaction.deserialize(raw_transaction)
            transaction.sign([payer])

            # blockhash
            bh_info = solana_client.get_latest_blockhash()
            transaction.recent_blockhash = bh_info['result']['value']['blockhash']

            # send
            result = solana_client.send_raw_transaction(transaction.serialize())
            print(f"Transaction sent: https://explorer.solana.com/tx/{result}")

        else:
            print(f"Error: {swap_response_data}")

    except Exception as e:
        print(f"Error during swap: {e}")

# Call Swap
swap_sol_to_token(amount_to_swap, "So11111111111111111111111111111111111111112", "3Buxeks8WjH9ermFa7a6AUrJcXX6EZ4Bb6xDUNzipump")

错误返回详情 #

No swapTransaction found in response.
Error: No swapTransaction found in response.
Quote Response Data: {
  "inputMint": "So11111111111111111111111111111111111111112",
  "inAmount": "10000000",
  "outputMint": "3Buxeks8WjH9ermFa7a6AUrJcXX6EZ4Bb6xDUNzipump",
  "outAmount": "44969051989",
  "otherAmountThreshold": "40472146791",
  "swapMode": "ExactIn",
  "slippageBps": 1000,
  "platformFee": null,
  "priceImpactPct": "0",
  "routePlan": [
    {
      "swapInfo": {
        "ammKey": "Dgt4UcEBzuR5f7VE2NrCyyQthEuWNzxsqf7pszSod1Sc",
        "label": "Raydium",
        "inputMint": "So11111111111111111111111111111111111111112",
        "outputMint": "3Buxeks8WjH9ermFa7a6AUrJcXX6EZ4Bb6xDUNzipump",
        "inAmount": "10000000",
        "outAmount": "44969051989",
        "feeAmount": "25000",
        "feeMint": "So11111111111111111111111111111111111111112"
      },
      "percent": 100
    }
  ],
  "scoreReport": null,
  "contextSlot": 304109036,
  "timeTaken": 0.000388887
}

解决方案 #

1. 理解Jupiter v6 API流程 #

Jupiter的v6 API分为两步：

• /quote接口：仅返回兑换报价、路由规划等信息，不会生成swapTransaction
• /swap接口：接收quote数据和用户公钥，生成可执行的swapTransaction

代码错误地在/quote的返回结果中检查swapTransaction，导致流程提前终止。

2. 修正代码逻辑 #

移除对quote数据中swapTransaction的检查，直接将quote数据传入/swap接口获取交易数据，同时修复其他潜在问题：

import requests
import base64
import re  # 新增缺失的re模块导入
from base58 import b58decode
from solana.rpc.api import Client
from solana.transaction import Transaction
from solders.keypair import Keypair
import json

# Solana Configuration
solana_client = Client("https://mainnet.helius-rpc.com/?api-key=API")
private_key_base58 = "KEY"
private_key_bytes = b58decode(private_key_base58)

# Jupiter API Configuration
JUPITER_API_URL = "https://quote-api.jup.ag/v6/quote"
SWAP_API_URL = "https://quote-api.jup.ag/v6/swap"

# Slippage
slippage = 0.1  # 10%

# Private Key 64 bytes
if len(private_key_bytes) == 64:
    payer = Keypair.from_bytes(private_key_bytes)
    print(f"Public Key: {payer.pubkey()}")
else:
    print("Error: Private key length is invalid. It should be 64 bytes.")

amount_to_swap = 0.01  # SOL

# ========== FUNCTIONALITY ==========

def extract_contract_address(message):
    pattern = r'\b[1-9A-HJ-NP-Za-km-z]{32,44}\b'
    match = re.search(pattern, message)
    if match:
        return match.group(0)
    return None

def get_quote(input_mint, output_mint, amount, slippage):
    params = {
        'inputMint': input_mint,
        'outputMint': output_mint,
        'amount': str(int(amount * 10**9)),
        'slippageBps': str(int(slippage * 10000))
    }

    response = requests.get(JUPITER_API_URL, params=params)
    if response.status_code == 200:
        data = response.json()
        print("Quote Data:", json.dumps(data, indent=2))
        # 移除对swapTransaction的检查，因为/quote接口不返回此字段
        return data
    else:
        print(f"Error fetching quote: {response.status_code}")
        return None

def swap_sol_to_token(input_amount, input_token_mint, output_token_mint):
    print(f"Swapping {input_amount} SOL for Token at Contract Address: {output_token_mint}...")

    try:
        quote_data = get_quote(input_token_mint, output_token_mint, input_amount, slippage)
        
        if not quote_data:
            print("Error: No valid data received from the quote API.")
            return

        # 移除对quote数据中swapTransaction的检查
        # 直接调用/swap接口获取交易数据
        swap_payload = {
            'quoteResponse': quote_data,
            'userPublicKey': str(payer.pubkey())
        }

        swap_response = requests.post(SWAP_API_URL, json=swap_payload)
        swap_response_data = swap_response.json()

        if swap_response.status_code == 200 and swap_response_data.get('swapTransaction'):
            swap_transaction = swap_response_data['swapTransaction']
            print(f"Swap Transaction: {swap_transaction}")

            # 修正交易处理顺序：先设置blockhash再签名
            raw_transaction = base64.b64decode(swap_transaction)
            transaction = Transaction.deserialize(raw_transaction)
            
            # 获取并设置最新blockhash
            bh_info = solana_client.get_latest_blockhash()
            transaction.recent_blockhash = bh_info['result']['value']['blockhash']
            
            # 签名交易
            transaction.sign([payer])

            # 发送交易
            result = solana_client.send_raw_transaction(transaction.serialize())
            print(f"Transaction sent: https://explorer.solana.com/tx/{result}")

        else:
            print(f"Error: {swap_response_data}")

    except Exception as e:
        print(f"Error during swap: {e}")

# Call Swap
swap_sol_to_token(amount_to_swap, "So11111111111111111111111111111111111111112", "3Buxeks8WjH9ermFa7a6AUrJcXX6EZ4Bb6xDUNzipump")

3. 关键修改点说明 #

• 移除get_quote和swap_sol_to_token中对quote数据的swapTransaction检查
• 新增缺失的re模块导入
• 调整交易处理顺序：先设置recent_blockhash再签名，避免签名失效

内容的提问来源于stack exchange，提问作者Narakorn Boonrasri