Python
Use full service RPC for fetching latest blockhash.
import aiohttp
import asyncio
import base64
from typing import List
from solders.pubkey import Pubkey
from solders.keypair import Keypair
from solders.instruction import Instruction
from solders.transaction import Transaction
from solders.system_program import transfer, TransferParams
from solders.hash import Hash
from solana.rpc.async_api import AsyncClient
NOZOMI_ENDPOINT = "https://nozomi.temporal.xyz/api/sendTransaction2?c=<YOUR_API_KEY>"
NOZOMI_TIP = Pubkey.from_string("TEMPaMeCRFAS9EKF53Jd6KpHxgL47uWLcpFArU1Fanq")
MIN_TIP_AMOUNT = 1_000_000
SOLANA_RPC_ENDPOINT = "https://api.mainnet-beta.solana.com"
async def send_nozomi_txn(
ixns: List[Instruction],
signer: Keypair,
nozomi_endpoint: str,
solana_rpc_client: AsyncClient
) -> None:
tip_ixn = transfer(TransferParams(
from_pubkey=signer.pubkey(),
to_pubkey=NOZOMI_TIP,
lamports=MIN_TIP_AMOUNT
))
ixns.append(tip_ixn)
blockhash_resp = await solana_rpc_client.get_latest_blockhash()
blockhash = blockhash_resp.value.blockhash
txn = Transaction.new_signed_with_payer(
ixns,
signer.pubkey(),
[signer],
blockhash
)
txn_bytes = bytes(txn)
txn_base64 = base64.b64encode(txn_bytes).decode('utf-8')
async with aiohttp.ClientSession() as session:
async with session.post(
nozomi_endpoint,
headers={"Content-Type": "text/plain"},
data=txn_base64
) as response:
# api v2 does not return a signature, just check for success
if response.status >= 200 and response.status < 300:
print("Transaction sent successfully")
else:
error_text = await response.text()
raise Exception(f"Transaction failed with status {response.status}: {error_text}")
def build_ixns() -> List[Instruction]:
# your instruction building logic here..
return []
async def main():
solana_rpc_client = AsyncClient(SOLANA_RPC_ENDPOINT)
# replace with actual keypair loading logic
signer = Keypair()
ixns = build_ixns()
await send_nozomi_txn(ixns, signer, NOZOMI_ENDPOINT, solana_rpc_client)
if __name__ == "__main__":
asyncio.run(main())Last updated

