Bundle = [Tx1, Tx2, ..., TxN]  (N <= 5)

- All transactions must be signed
- Transactions execute in order: Tx1 → Tx2 → ... → TxN
- If ANY transaction fails, the ENTIRE bundle is dropped
- The tip instruction goes in the LAST transaction (last instruction)
- Bundle has ~2 slots (~800ms) to land before expiry

undefined

Tip accounts are fetched dynamically via `getTipAccounts`. Rotate through them to distribute load.

import httpx

BLOCK_ENGINE = "https://mainnet.block-engine.jito.wtf"

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "sendBundle",
    "params": [
        [tx1_base58, tx2_base58],  # List of base58-encoded signed txs
    ]
}

resp = httpx.post(f"{BLOCK_ENGINE}/api/v1/bundles", json=payload)
data = resp.json()
bundle_id = data["result"]  # UUID string

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "getBundleStatuses",
    "params": [[bundle_id]]
}
resp = httpx.post(f"{BLOCK_ENGINE}/api/v1/bundles", json=payload)
statuses = resp.json()["result"]["value"]

undefined

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "getTipAccounts",
    "params": []
}
resp = httpx.post(f"{BLOCK_ENGINE}/api/v1/bundles", json=payload)
tip_accounts = resp.json()["result"]  # List of 8 base58 pubkeys

payload = {
    "jsonrpc": "2.0",
    "id": 1,
    "method": "getInflightBundleStatuses",
    "params": [[bundle_id]]
}
resp = httpx.post(f"{BLOCK_ENGINE}/api/v1/bundles", json=payload)

undefined

from solders.transaction import VersionedTransaction
from solders.message import MessageV0
from solders.instruction import Instruction
from solders.system_program import transfer, TransferParams
from solders.pubkey import Pubkey
import random

def build_protected_swap_bundle(
    swap_ix: Instruction,
    payer: Pubkey,
    tip_lamports: int,
    tip_accounts: list[str],
    recent_blockhash: str,
) -> list[VersionedTransaction]:
    """Build a 1-tx bundle: swap + tip in the same transaction.

    For simple swaps, a single-transaction bundle is sufficient.
    The tip instruction is appended as the last instruction.
    """
    # Pick a random tip account
    tip_account = Pubkey.from_string(random.choice(tip_accounts))

    # Tip instruction
    tip_ix = transfer(TransferParams(
        from_pubkey=payer,
        to_pubkey=tip_account,
        lamports=tip_lamports,
    ))

    # Build transaction with swap + tip
    msg = MessageV0.try_compile(
        payer=payer,
        instructions=[swap_ix, tip_ix],
        address_lookup_table_accounts=[],
        recent_blockhash=recent_blockhash,
    )
    tx = VersionedTransaction(msg, [keypair])
    return [tx]

def calculate_dynamic_tip(
    base_tip: int = 10_000,
    urgency_multiplier: float = 1.0,
    recent_tip_percentile_50: int = 15_000,
) -> int:
    """Calculate tip based on urgency and recent network tips.

    Args:
        base_tip: Minimum tip in lamports.
        urgency_multiplier: 1.0 = normal, 2.0 = urgent, 5.0 = critical.
        recent_tip_percentile_50: Median tip from recent bundles.

    Returns:
        Tip amount in lamports.
    """
    dynamic_tip = max(base_tip, int(recent_tip_percentile_50 * urgency_multiplier))
    # Cap at 0.01 SOL to prevent accidents
    return min(dynamic_tip, 10_000_000)

async def submit_with_retry(
    bundle_txs: list[str],
    endpoints: list[str],
    max_retries: int = 3,
) -> str | None:
    """Submit bundle to multiple endpoints with retry logic.

    Returns bundle_id if submitted, None if all retries exhausted.
    """
    for attempt in range(max_retries):
        # Submit to all endpoints in parallel
        async with httpx.AsyncClient() as client:
            tasks = [
                client.post(
                    f"{ep}/api/v1/bundles",
                    json={
                        "jsonrpc": "2.0", "id": 1,
                        "method": "sendBundle",
                        "params": [bundle_txs],
                    },
                    timeout=5.0,
                )
                for ep in endpoints
            ]
            # Process first successful response
            for resp in asyncio.as_completed(tasks):
                result = (await resp).json()
                if "result" in result:
                    return result["result"]

        # Wait before retry with fresh blockhash
        await asyncio.sleep(0.5 * (attempt + 1))
    return None