python

Send a bundle using Python.

Using Requests

import requests
import base64
from typing import List

ENDPOINT = "https://fra.blocksprint.io/bundle/v1/YOUR_API_KEY"

def send_bundle(transactions: List[bytes]) -> dict:
    """
    Send a bundle of transactions.

    Args:
        transactions: List of serialized transaction bytes

    Returns:
        Bundle result with bundle_id and signatures
    """
    # Encode transactions to base64
    encoded_txs = [
        base64.b64encode(tx).decode("utf-8")
        for tx in transactions
    ]

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "sendBundle",
        "params": [
            encoded_txs,
            {
                "encoding": "base64",
                "skipPreflight": True,
            }
        ]
    }

    response = requests.post(
        ENDPOINT,
        json=payload,
        headers={"Content-Type": "application/json"}
    )

    result = response.json()

    if "error" in result:
        raise Exception(f"Bundle error: {result['error']['message']}")

    return result["result"]

Full Example with solders

from solders.hash import Hash
from solders.keypair import Keypair
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction
import requests
import base64

ENDPOINT = "https://fra.blocksprint.io/bundle/v1/YOUR_API_KEY"
RPC_URL = "https://api.mainnet-beta.solana.com"  # Your Solana RPC (for the blockhash)
TIP_ADDRESS = Pubkey.from_string("Sp4JHSh9cksfzXbgK7Pq2ovtn8LirLQydaJKTsiNT77")  # BlockSprint tip address
TIP_LAMPORTS = 1_000_000  # 0.001 SOL

def get_latest_blockhash() -> Hash:
    response = requests.post(
        RPC_URL,
        json={"jsonrpc": "2.0", "id": 1, "method": "getLatestBlockhash"}
    ).json()
    return Hash.from_string(response["result"]["value"]["blockhash"])

def create_bundle_transactions(wallet: Keypair, recipients: list, amounts: list):
    """Create multiple transactions for a bundle."""
    blockhash = get_latest_blockhash()

    transactions = []

    for recipient, amount in zip(recipients, amounts):
        tx = Transaction.new_signed_with_payer(
            [
                transfer(
                    TransferParams(
                        from_pubkey=wallet.pubkey(),
                        to_pubkey=Pubkey.from_string(recipient),
                        lamports=amount,
                    )
                )
            ],
            wallet.pubkey(),
            [wallet],
            blockhash,
        )
        transactions.append(bytes(tx))

    # Tip transaction (0.001 SOL to a BlockSprint tip address)
    tip_tx = Transaction.new_signed_with_payer(
        [
            transfer(
                TransferParams(
                    from_pubkey=wallet.pubkey(),
                    to_pubkey=TIP_ADDRESS,
                    lamports=TIP_LAMPORTS,
                )
            )
        ],
        wallet.pubkey(),
        [wallet],
        blockhash,
    )
    transactions.append(bytes(tip_tx))

    return transactions

def execute_bundle(wallet: Keypair, recipients: list, amounts: list):
    """Create and send a bundle."""
    transactions = create_bundle_transactions(wallet, recipients, amounts)
    result = send_bundle(transactions)

    print(f"Bundle ID: {result['bundle_id']}")
    for i, sig in enumerate(result['signatures']):
        print(f"Tx {i + 1}: {sig}")

    return result

# Usage
wallet = Keypair()  # Your wallet
recipients = ["ADDRESS_1", "ADDRESS_2"]
amounts = [1_000_000, 2_000_000]

result = execute_bundle(wallet, recipients, amounts)

Async Version

import aiohttp
import asyncio
import base64
from typing import List

ENDPOINT = "https://fra.blocksprint.io/bundle/v1/YOUR_API_KEY"

async def send_bundle_async(transactions: List[bytes]) -> dict:
    """Send a bundle asynchronously."""
    encoded_txs = [
        base64.b64encode(tx).decode("utf-8")
        for tx in transactions
    ]

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "sendBundle",
        "params": [
            encoded_txs,
            {
                "encoding": "base64",
                "skipPreflight": True,
            }
        ]
    }

    async with aiohttp.ClientSession() as session:
        async with session.post(ENDPOINT, json=payload) as response:
            result = await response.json()

    if "error" in result:
        raise Exception(f"Bundle error: {result['error']['message']}")

    return result["result"]

# Usage
async def main():
    transactions = [...]  # Your serialized transactions
    result = await send_bundle_async(transactions)
    print(result)

asyncio.run(main())

Arbitrage Example

def execute_arbitrage(wallet: Keypair, dex_a, dex_b, token, amount):
    """Execute an arbitrage bundle."""
    blockhash = get_latest_blockhash()

    # Transaction 1: Buy on DEX A
    buy_tx = Transaction.new_signed_with_payer(
        [dex_a.create_buy_instruction(wallet.pubkey(), token, amount)],
        wallet.pubkey(),
        [wallet],
        blockhash,
    )

    # Transaction 2: Sell on DEX B, plus the 0.001 SOL tip
    sell_tx = Transaction.new_signed_with_payer(
        [
            dex_b.create_sell_instruction(wallet.pubkey(), token, amount),
            transfer(
                TransferParams(
                    from_pubkey=wallet.pubkey(),
                    to_pubkey=TIP_ADDRESS,
                    lamports=TIP_LAMPORTS,
                )
            ),
        ],
        wallet.pubkey(),
        [wallet],
        blockhash,
    )

    # Send as atomic bundle
    result = send_bundle([bytes(buy_tx), bytes(sell_tx)])

    return result