Skip to content

python ​

Send a transaction using Python with solana-py or raw HTTP.

Using solana-py ​

bash
pip install solana
python
import asyncio
import os
import random

from solana.rpc.async_api import AsyncClient
from solana.rpc.models import TxOpts
from solders.keypair import Keypair
from solders.message import Message
from solders.pubkey import Pubkey
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction

# Your Solana RPC: blockhash and confirmations
RPC_URL = "https://api.mainnet-beta.solana.com"
# BlockSprint: sending only
ENDPOINT = "https://fra.blocksprint.io/api/v1/YOUR_API_KEY"

# BlockSprint tip addresses
TIP_ADDRESSES = [
    "Sp1x2AqpQckPLaWnWCJUNg8k6qQexfaEWcSRKf5JcDV",
    "Sp4JHSh9cksfzXbgK7Pq2ovtn8LirLQydaJKTsiNT77",
    "Sp1xMS2cbw83SZDNr4AGqkBYYLjb3LvVnmDSrTMaHkr",
    "SpagSJmnh8E9cGT5Y431xPPaS2c1xLREGGCWN9yDeUf",
    "SpWrza9E63MQuHeGnnfzmtLVCs3pBdjyKPXUABPo9nq",
    "SpXyoqMHJEhsNQ9fHh3DTGGzrGroZ5zfDtJwFvN4kXi",
    "Spv2g3L837bVeHwkyb18LhA5P1bvJbkvaZbxcZXbwC2",
    "SpDUe5MyFGG97SmdBQMWNPbx2Et9bZt2TCbZm3TujgX",
    "SpKZ28tFzQ13kVjHhsyDEiQfkPGjA7Wmk185KcEDyuh",
    "SpxPqegfoR3QkBN7guRUiU6QfhtTW1CnKSTSSapumqm",
]

def get_random_tip_address() -> Pubkey:
    return Pubkey.from_string(random.choice(TIP_ADDRESSES))

async def send_transaction(wallet: Keypair, recipient: Pubkey, amount: int) -> str:
    async with AsyncClient(RPC_URL) as rpc, AsyncClient(ENDPOINT) as blocksprint:
        instructions = [
            # Your transfer
            transfer(
                TransferParams(
                    from_pubkey=wallet.pubkey(),
                    to_pubkey=recipient,
                    lamports=amount,
                )
            ),
            # Tip instruction (0.001 SOL minimum)
            transfer(
                TransferParams(
                    from_pubkey=wallet.pubkey(),
                    to_pubkey=get_random_tip_address(),
                    lamports=1_000_000,
                )
            ),
        ]

        # Get blockhash from your RPC
        blockhash = (await rpc.get_latest_blockhash()).value.blockhash

        # Build and sign
        message = Message.new_with_blockhash(instructions, wallet.pubkey(), blockhash)
        transaction = Transaction([wallet], message, blockhash)

        # Send through BlockSprint (solana-py sends base64)
        result = await blocksprint.send_raw_transaction(
            bytes(transaction),
            opts=TxOpts(skip_preflight=True, max_retries=0),
        )

        # "Accepted", not "landed": check it with rpc.get_signature_statuses([...])
        return str(result.value)

# Usage
wallet = Keypair.from_base58_string(os.environ["PAYER_SECRET_BASE58"])  # your funded wallet
recipient = Pubkey.from_string(os.environ["RECIPIENT"])
signature = asyncio.run(send_transaction(wallet, recipient, 1_000_000))
print(f"Signature: {signature}")

Using Raw HTTP ​

python
import requests
import base64

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

# Reuse one session so the connection to BlockSprint stays open between sends
session = requests.Session()

def send_transaction_raw(transaction_bytes: bytes) -> str:
    base64_tx = base64.b64encode(transaction_bytes).decode("utf-8")

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "sendTransaction",
        "params": [
            base64_tx,
            {
                "encoding": "base64",
                "skipPreflight": True,
                "maxRetries": 0,
            }
        ]
    }

    response = session.post(ENDPOINT, json=payload, timeout=10)

    # Errors come back as JSON-RPC errors with an HTTP 4xx/5xx status
    result = response.json()

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

    return result["result"]

# Usage
serialized_tx = bytes(transaction)
signature = send_transaction_raw(serialized_tx)
print(f"Signature: {signature}")

With Priority Fee ​

python
from solders.compute_budget import set_compute_unit_limit, set_compute_unit_price

instructions = [
    # Priority fee (micro-lamports per compute unit)
    set_compute_unit_price(1_000_000),
    set_compute_unit_limit(200_000),
    # Your instruction
    your_instruction,
    # Tip
    tip_instruction,
]

Async Version ​

python
import aiohttp
import asyncio
import base64

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

async def send_transaction_async(transaction_bytes: bytes) -> str:
    base64_tx = base64.b64encode(transaction_bytes).decode("utf-8")

    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "sendTransaction",
        "params": [
            base64_tx,
            {
                "encoding": "base64",
                "skipPreflight": True,
                "maxRetries": 0,
            }
        ]
    }

    # In production, create one ClientSession and reuse it for every send
    async with aiohttp.ClientSession() as session:
        async with session.post(ENDPOINT, json=payload) as response:
            # content_type=None: also parse the JSON body of 4xx error responses
            result = await response.json(content_type=None)

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

    return result["result"]

# Usage
signature = asyncio.run(send_transaction_async(serialized_tx))

INFO

Transaction V1 needs solders 0.29.0 or later: compile the message with MessageV1.try_compile and a TransactionConfig that sets the compute-unit limit and the loaded-accounts data size limit (in V1 an unset limit is 0). bytes(transaction) is the wire format to send.