import asyncio
import base64
import httpx
from solders.commitment_config import CommitmentLevel
from solders.hash import Hash
from solders.keypair import Keypair
from solders.message import Message
from solders.pubkey import Pubkey
from solders.rpc.config import RpcContextConfig, RpcSendTransactionConfig
from solders.rpc.requests import GetLatestBlockhash, SendVersionedTransaction
from solders.system_program import TransferParams, transfer
from solders.transaction import Transaction, VersionedTransaction
async def buy_token_local(mint: Pubkey):
async with httpx.AsyncClient() as client:
try:
# Get the create transaction instruction
response = await client.post(
"https://api.soltool.pro/rpc",
json={
"jsonrpc": "2.0",
"id": 1,
"method": "soltool_getTradeInstruction",
"params": {
"request": {
"publicKey": "<YOUR_PUBLIC_KEY>",
"action": "Buy",
"mint": str(mint),
"amount": 1000000,
"slippage": 100, # 1%
"priorityFee": 1000,
"pool": "PumpFun",
}
},
},
)
response.raise_for_status()
result = response.json()
if result.get("result", None) is not None:
# Get transaction instruction
decoded_data = base64.b64decode(result["result"])
# Create transaction
signer_keypair = Keypair.from_base58_string("<YOUR_PRIVATE_KEY>")
tx = VersionedTransaction(
VersionedTransaction.from_bytes(decoded_data).message,
[signer_keypair],
)
# Send transaction
commitment = CommitmentLevel.Confirmed
config = RpcSendTransactionConfig(preflight_commitment=commitment)
tx_payload = SendVersionedTransaction(tx, config)
response = await client.post(
url="https://api.mainnet-beta.solana.com/",
headers={"Content-Type": "application/json"},
data=tx_payload.to_json(),
)
response.raise_for_status()
result = response.json()
if "result" in result:
tx_signature = result["result"]
print(f"Transaction: https://solscan.io/tx/{tx_signature}")
else:
print(
"Error:",
result.get("error", {}).get("message", "Unknown error"),
)
print(
"Details:", result.get("error", {}).get("data", "Unknown error")
)
else:
print("Error:", result.get("error", {}).get("message", "Unknown error"))
print("Details:", result.get("error", {}).get("data", "Unknown error"))
except httpx.HTTPError as e:
print(f"HTTP error occurred: {e}")
except Exception as e:
print(f"An error occurred: {e}")