Solana Rpc Agent Skill

import httpx

RPC = os.getenv("SOLANA_RPC_URL", "https://api.mainnet-beta.solana.com")

def rpc_call(method: str, params: list = None) -> dict:
    resp = httpx.post(RPC, json={
        "jsonrpc": "2.0", "id": 1,
        "method": method, "params": params or [],
    }, timeout=30.0)
    return resp.json()

# Get SOL balance
result = rpc_call("getBalance", ["WALLET_PUBKEY"])
sol_balance = result["result"]["value"] / 1e9

# Get latest blockhash
result = rpc_call("getLatestBlockhash")
blockhash = result["result"]["value"]["blockhash"]

# SOL balance (in lamports, divide by 1e9 for SOL)
getBalance(pubkey, {commitment: "confirmed"})

# Full account info (data, owner, lamports, executable)
getAccountInfo(pubkey, {encoding: "jsonParsed"})

# Multiple accounts in one call
getMultipleAccounts([pubkey1, pubkey2], {encoding: "jsonParsed"})

# All SPL token accounts owned by a wallet
getTokenAccountsByOwner(wallet_pubkey, {
    "programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"
}, {"encoding": "jsonParsed"})

# Token balance for a specific token account
getTokenAccountBalance(token_account_pubkey)

# Largest token accounts (top holders)
getTokenLargestAccounts(mint_pubkey)

# Total supply of a token
getTokenSupply(mint_pubkey)

# Get parsed transaction by signature
getTransaction(signature, {
    "encoding": "jsonParsed",
    "maxSupportedTransactionVersion": 0,
})

# Recent transaction signatures for an address
getSignaturesForAddress(pubkey, {
    "limit": 20,
    "before": "optional_signature",  # pagination cursor
})

# Transaction status
getSignatureStatuses([sig1, sig2])

# Current slot
getSlot({commitment: "confirmed"})

# Block data
getBlock(slot, {
    "encoding": "jsonParsed",
    "transactionDetails": "full",
    "maxSupportedTransactionVersion": 0,
})

# Latest blockhash (needed for tx building)
getLatestBlockhash({commitment: "confirmed"})

# Slot leader schedule
getLeaderSchedule()

# All accounts owned by a program (with filters)
getProgramAccounts(program_pubkey, {
    "encoding": "jsonParsed",
    "filters": [
        {"dataSize": 165},  # Filter by account data size
        {"memcmp": {         # Filter by data content
            "offset": 32,
            "bytes": "base58_encoded_value",
        }},
    ],
})

# Recent priority fee estimates
getRecentPrioritizationFees([account_pubkey])
# Returns array of { slot, prioritizationFee } for recent slots

# Minimum rent for account
getMinimumBalanceForRentExemption(data_length)

# Send a signed, serialized transaction
sendTransaction(base64_tx, {
    "encoding": "base64",
    "skipPreflight": False,
    "preflightCommitment": "confirmed",
    "maxRetries": 3,
})

# Simulate before sending
simulateTransaction(base64_tx, {
    "encoding": "base64",
    "sigVerify": False,
    "commitment": "confirmed",
})

import time

def confirm_transaction(rpc_url: str, signature: str, timeout: float = 30.0) -> bool:
    """Poll for transaction confirmation."""
    start = time.time()
    while time.time() - start < timeout:
        result = rpc_call("getSignatureStatuses", [[signature]])
        statuses = result.get("result", {}).get("value", [None])
        if statuses[0] is not None:
            status = statuses[0]
            if status.get("err"):
                return False
            if status.get("confirmationStatus") in ("confirmed", "finalized"):
                return True
        time.sleep(0.5)
    return False

def get_wallet_tokens(wallet: str) -> list[dict]:
    """Get all SPL token holdings with metadata."""
    result = rpc_call("getTokenAccountsByOwner", [
        wallet,
        {"programId": "TokenkegQfeZyiNwAJbNbGKPFXCWuBvf9Ss623VQ5DA"},
        {"encoding": "jsonParsed"},
    ])
    tokens = []
    for acct in result["result"]["value"]:
        info = acct["account"]["data"]["parsed"]["info"]
        tokens.append({
            "mint": info["mint"],
            "amount": int(info["tokenAmount"]["amount"]),
            "decimals": info["tokenAmount"]["decimals"],
            "ui_amount": info["tokenAmount"]["uiAmount"],
        })
    return [t for t in tokens if t["amount"] > 0]

def get_top_holders(mint: str) -> list[dict]:
    """Get the 20 largest holders of a token."""
    result = rpc_call("getTokenLargestAccounts", [mint])
    supply_result = rpc_call("getTokenSupply", [mint])
    total_supply = int(supply_result["result"]["value"]["amount"])

    holders = []
    for acct in result["result"]["value"]:
        amount = int(acct["amount"])
        holders.append({
            "address": acct["address"],
            "amount": amount,
            "decimals": acct["decimals"],
            "ui_amount": acct["uiAmount"],
            "percentage": amount / total_supply * 100 if total_supply > 0 else 0,
        })
    return holders

def rpc_batch(calls: list[tuple[str, list]]) -> list[dict]:
    """Execute multiple RPC calls in a single HTTP request."""
    payload = [
        {"jsonrpc": "2.0", "id": i, "method": method, "params": params}
        for i, (method, params) in enumerate(calls)
    ]
    resp = httpx.post(RPC, json=payload, timeout=30.0)
    results = resp.json()
    results.sort(key=lambda r: r["id"])
    return [r.get("result") for r in results]