Model Context Protocol Server for Solana Client

  • src
from typing import Optional from mcp.server import FastMCP from solana.rpc.async_api import AsyncClient from solana.rpc.commitment import Commitment from solana.rpc.types import TokenAccountOpts from solders.message import Message # type: ignore from solders.pubkey import Pubkey # type: ignore from solders.signature import Signature # type: ignore from solders.system_program import TransferParams, transfer mcp = FastMCP("Solana Client") rpc_url = "https://api.mainnet-beta.solana.com" @mcp.tool() async def get_balance(address: str) -> str: """Returns the balance of the account of provided Pubkey. Args: address (str): Pubkey of account to query Returns: str: Account balance response in the format "Balance of {address}: {balance}" """ async with AsyncClient(rpc_url) as client: balance = await client.get_balance(Pubkey.from_string(address)) return f"Balance of {address}: {balance}" @mcp.tool() async def get_transaction(hash: str) -> str: """Returns transaction details for a confirmed transaction. Args: hash (str): Transaction signature as base-58 encoded string Returns: str: Transaction details in the format "Transaction: {transaction}" """ async with AsyncClient(rpc_url) as client: transaction = await client.get_transaction( Signature.from_string(hash), max_supported_transaction_version=0 ) return f"Transaction: {transaction}" @mcp.tool() async def get_block(slot: int) -> str: """Returns identity and transaction information about a confirmed block in the ledger. Args: slot (int): Slot number as u64 integer Returns: str: Block information in the format "Block: {block}" """ async with AsyncClient(rpc_url) as client: block = await client.get_block(slot) return f"Block: {block}" @mcp.tool() async def get_block_height() -> str: """Returns the current block height of the node. Returns: str: Current block height in the format "Block height: {height}" """ async with AsyncClient(rpc_url) as client: height = await client.get_block_height() return f"Block height: {height}" @mcp.tool() async def get_block_time(slot: int) -> str: """Fetch the estimated production time of a block. Args: slot (int): Block slot number Returns: str: Block time in the format "Block time: {time}" """ async with AsyncClient(rpc_url) as client: time = await client.get_block_time(slot) return f"Block time: {time}" @mcp.tool() async def get_blocks(start_slot: int, end_slot: Optional[int] = None) -> str: """Returns a list of confirmed blocks between two slots. Args: start_slot (int): Start slot as u64 integer end_slot (Optional[int], optional): End slot as u64 integer. Defaults to None. Returns: str: List of blocks in the format "Blocks: {blocks}" """ async with AsyncClient(rpc_url) as client: blocks = await client.get_blocks(start_slot, end_slot) return f"Blocks: {blocks}" @mcp.tool() async def get_cluster_nodes() -> str: """Returns information about all the nodes participating in the cluster. Returns: str: Cluster nodes information in the format "Cluster nodes: {nodes}" """ async with AsyncClient(rpc_url) as client: nodes = await client.get_cluster_nodes() return f"Cluster nodes: {nodes}" @mcp.tool() async def get_epoch_info() -> str: """Returns information about the current epoch. Returns: str: Epoch information in the format "Epoch info: {info}" """ async with AsyncClient(rpc_url) as client: info = await client.get_epoch_info() return f"Epoch info: {info}" @mcp.tool() async def get_epoch_schedule() -> str: """Returns epoch schedule information from this cluster's genesis config. Returns: str: Epoch schedule in the format "Epoch schedule: {schedule}" """ async with AsyncClient(rpc_url) as client: schedule = await client.get_epoch_schedule() return f"Epoch schedule: {schedule}" @mcp.tool() async def get_genesis_hash() -> str: """Returns the genesis hash. Returns: str: Genesis hash in the format "Genesis hash: {hash}" """ async with AsyncClient(rpc_url) as client: hash = await client.get_genesis_hash() return f"Genesis hash: {hash}" @mcp.tool() async def get_identity() -> str: """Returns the identity pubkey for the current node. Returns: str: Node identity in the format "Node identity: {identity}" """ async with AsyncClient(rpc_url) as client: identity = await client.get_identity() return f"Node identity: {identity}" @mcp.tool() async def get_inflation_governor() -> str: """Returns the current inflation governor. Returns: str: Inflation governor info in the format "Inflation governor: {governor}" """ async with AsyncClient(rpc_url) as client: governor = await client.get_inflation_governor() return f"Inflation governor: {governor}" @mcp.tool() async def get_inflation_rate() -> str: """Returns the specific inflation values for the current epoch. Returns: str: Inflation rate info in the format "Inflation rate: {rate}" """ async with AsyncClient(rpc_url) as client: rate = await client.get_inflation_rate() return f"Inflation rate: {rate}" @mcp.tool() async def get_largest_accounts() -> str: """Returns the 20 largest accounts, by lamport balance. Returns: str: Largest accounts info in the format "Largest accounts: {accounts}" """ async with AsyncClient(rpc_url) as client: accounts = await client.get_largest_accounts() return f"Largest accounts: {accounts}" @mcp.tool() async def get_latest_blockhash() -> str: """Returns the latest block hash from the ledger. Returns: str: Latest blockhash in the format "Latest blockhash: {blockhash}" """ async with AsyncClient(rpc_url) as client: blockhash = await client.get_latest_blockhash() return f"Latest blockhash: {blockhash}" @mcp.tool() async def get_minimum_balance_for_rent_exemption(size: int) -> str: """Returns minimum balance required to make account rent exempt. Args: size (int): Account data length Returns: str: Minimum balance in the format "Minimum balance for rent exemption: {balance}" """ async with AsyncClient(rpc_url) as client: balance = await client.get_minimum_balance_for_rent_exemption(size) return f"Minimum balance for rent exemption: {balance}" @mcp.tool() async def get_program_accounts(program_id: str) -> str: """Returns all accounts owned by the provided program Pubkey. Args: program_id (str): Pubkey of program to query Returns: str: Program accounts in the format "Program accounts: {accounts}" """ async with AsyncClient(rpc_url) as client: accounts = await client.get_program_accounts(Pubkey.from_string(program_id)) return f"Program accounts: {accounts}" @mcp.tool() async def get_recent_performance_samples(limit: Optional[int] = None) -> str: """Returns a list of recent performance samples, in reverse slot order. Args: limit (Optional[int], optional): Number of samples to return (maximum 720). Defaults to None. Returns: str: Performance samples in the format "Performance samples: {samples}" """ async with AsyncClient(rpc_url) as client: samples = await client.get_recent_performance_samples(limit) return f"Performance samples: {samples}" @mcp.tool() async def get_signature_statuses(signatures: list[str]) -> str: """Returns the statuses of a list of signatures. Args: signatures (list[str]): List of transaction signatures to confirm Returns: str: Signature statuses in the format "Signature statuses: {statuses}" """ async with AsyncClient(rpc_url) as client: sigs = [Signature.from_string(sig) for sig in signatures] statuses = await client.get_signature_statuses(sigs) return f"Signature statuses: {statuses}" @mcp.tool() async def get_slot() -> str: """Returns the current slot the node is processing. Returns: str: Current slot in the format "Current slot: {slot}" """ async with AsyncClient(rpc_url) as client: slot = await client.get_slot() return f"Current slot: {slot}" @mcp.tool() async def get_slot_leader() -> str: """Returns the current slot leader. Returns: str: Slot leader in the format "Slot leader: {leader}" """ async with AsyncClient(rpc_url) as client: leader = await client.get_slot_leader() return f"Slot leader: {leader}" @mcp.tool() async def get_supply() -> str: """Returns information about the current supply. Returns: str: Supply information in the format "Supply info: {supply}" """ async with AsyncClient(rpc_url) as client: supply = await client.get_supply() return f"Supply info: {supply}" @mcp.tool() async def get_token_account_balance(token_account: str) -> str: """Returns the token balance of an SPL Token account. Args: token_account (str): Pubkey of Token account to query Returns: str: Token account balance in the format "Token account balance: {balance}" """ async with AsyncClient(rpc_url) as client: balance = await client.get_token_account_balance( Pubkey.from_string(token_account) ) return f"Token account balance: {balance}" @mcp.tool() async def get_token_largest_accounts(mint: str) -> str: """Returns the 20 largest accounts of a particular SPL Token type. Args: mint (str): Pubkey of token mint to query Returns: str: Largest token accounts in the format "Largest token accounts: {accounts}" """ async with AsyncClient(rpc_url) as client: accounts = await client.get_token_largest_accounts(Pubkey.from_string(mint)) return f"Largest token accounts: {accounts}" @mcp.tool() async def get_transaction_count() -> str: """Returns the current Transaction count from the ledger. Returns: str: Transaction count in the format "Transaction count: {count}" """ async with AsyncClient(rpc_url) as client: count = await client.get_transaction_count() return f"Transaction count: {count}" @mcp.tool() async def get_version() -> str: """Returns the current solana versions running on the node. Returns: str: Version information in the format "Version info: {version}" """ async with AsyncClient(rpc_url) as client: version = await client.get_version() return f"Version info: {version}" @mcp.tool() async def get_vote_accounts() -> str: """Returns the account info and associated stake for all the voting accounts in the current bank. Returns: str: Vote accounts information in the format "Vote accounts: {accounts}" """ async with AsyncClient(rpc_url) as client: accounts = await client.get_vote_accounts() return f"Vote accounts: {accounts}" @mcp.tool() async def is_connected() -> str: """Health check to verify if the client is connected. Returns: str: Connection status in the format "Connected: {connected}" """ async with AsyncClient(rpc_url) as client: connected = await client.is_connected() return f"Connected: {connected}" @mcp.tool() async def get_block_commitment(slot: int) -> str: """Fetch the commitment for particular block. Args: slot (int): Block slot number to query Returns: str: Block commitment information """ async with AsyncClient(rpc_url) as client: commitment = await client.get_block_commitment(slot) return f"Block commitment: {commitment}" @mcp.tool() async def confirm_transaction(tx_sig: str, commitment: Optional[str] = None) -> str: """Confirm the transaction identified by the specified signature. Args: tx_sig (str): Transaction signature to confirm commitment (Optional[str]): Bank state to query ("finalized", "confirmed" or "processed") Returns: str: Transaction confirmation status """ async with AsyncClient(rpc_url) as client: result = await client.confirm_transaction( Signature.from_string(tx_sig), Commitment(commitment) if commitment else None, ) return f"Transaction confirmation: {result}" @mcp.tool() async def get_account_info(pubkey: str, encoding: str = "base64") -> str: """Returns all account info for the specified public key. Args: pubkey (str): Pubkey of account to query encoding (str): Encoding for Account data ("base58", "base64", or "jsonParsed") Returns: str: Account information """ async with AsyncClient(rpc_url) as client: info = await client.get_account_info( Pubkey.from_string(pubkey), encoding=encoding ) return f"Account info: {info}" @mcp.tool() async def get_fee_for_message(from_pubkey: str, to_pubkey: str, lamports: int) -> str: """Returns the fee for a message. Args: from_pubkey (str): Sender's public key to_pubkey (str): Recipient's public key lamports (int): Amount of lamports to transfer Returns: str: Fee information """ async with AsyncClient(rpc_url) as client: msg = Message( [ transfer( TransferParams( from_pubkey=Pubkey.from_string(from_pubkey), to_pubkey=Pubkey.from_string(to_pubkey), lamports=lamports, ) ) ] ) fee = await client.get_fee_for_message(msg) return f"Message fee: {fee}" @mcp.tool() async def get_first_available_block() -> str: """Returns the slot of the lowest confirmed block available. Returns: str: First available block information """ async with AsyncClient(rpc_url) as client: block = await client.get_first_available_block() return f"First available block: {block}" @mcp.tool() async def get_inflation_reward(pubkeys: list[str], epoch: Optional[int] = None) -> str: """Returns the inflation/staking reward for a list of addresses for an epoch. Args: pubkeys (list[str]): List of account addresses epoch (Optional[int]): Epoch for which to calculate rewards Returns: str: Inflation reward information """ async with AsyncClient(rpc_url) as client: pks = [Pubkey.from_string(pk) for pk in pubkeys] rewards = await client.get_inflation_reward(pks, epoch) return f"Inflation rewards: {rewards}" @mcp.tool() async def get_leader_schedule(epoch: Optional[int] = None) -> str: """Returns the leader schedule for an epoch. Args: epoch (Optional[int]): Epoch to get schedule for Returns: str: Leader schedule information """ async with AsyncClient(rpc_url) as client: schedule = await client.get_leader_schedule(epoch) return f"Leader schedule: {schedule}" @mcp.tool() async def get_minimum_ledger_slot() -> str: """Returns the lowest slot that the node has information about in its ledger. Returns: str: Minimum ledger slot information """ async with AsyncClient(rpc_url) as client: slot = await client.get_minimum_ledger_slot() return f"Minimum ledger slot: {slot}" @mcp.tool() async def get_multiple_accounts(pubkeys: list[str], encoding: str = "base64") -> str: """Returns the account information for a list of public keys. Args: pubkeys (list[str]): List of account public keys encoding (str): Encoding for the account data Returns: str: Multiple accounts information """ async with AsyncClient(rpc_url) as client: pks = [Pubkey.from_string(pk) for pk in pubkeys] accounts = await client.get_multiple_accounts(pks, encoding=encoding) return f"Multiple accounts info: {accounts}" @mcp.tool() async def get_signatures_for_address( account: str, before: Optional[str] = None, until: Optional[str] = None, limit: Optional[int] = None, ) -> str: """Returns confirmed signatures for transactions involving an address. Args: account (str): Account address to query before (Optional[str]): Start searching backwards from this signature until (Optional[str]): Search until this signature limit (Optional[int]): Maximum number of signatures to return Returns: str: Signatures information """ async with AsyncClient(rpc_url) as client: sigs = await client.get_signatures_for_address( Pubkey.from_string(account), before=Signature.from_string(before) if before else None, until=Signature.from_string(until) if until else None, limit=limit, ) return f"Signatures for address: {sigs}" @mcp.tool() async def get_token_accounts_by_delegate(delegate: str, mint: str) -> str: """Returns all SPL Token accounts by approved delegate. Args: delegate (str): Public key of delegate owner mint (str): Token mint address Returns: str: Token accounts information """ async with AsyncClient(rpc_url) as client: accounts = await client.get_token_accounts_by_delegate( Pubkey.from_string(delegate), TokenAccountOpts(Pubkey.from_string(mint)) ) return f"Token accounts by delegate: {accounts}" @mcp.tool() async def get_token_accounts_by_owner(owner: str, mint: str) -> str: """Returns all SPL Token accounts by token owner. Args: owner (str): Public key of token owner mint (str): Token mint address Returns: str: Token accounts information """ async with AsyncClient(rpc_url) as client: accounts = await client.get_token_accounts_by_owner( Pubkey.from_string(owner), TokenAccountOpts(Pubkey.from_string(mint)) ) return f"Token accounts by owner: {accounts}" @mcp.tool() async def get_token_supply(mint: str) -> str: """Returns the total supply of an SPL Token type. Args: mint (str): Public key of token mint Returns: str: Token supply information """ async with AsyncClient(rpc_url) as client: supply = await client.get_token_supply(Pubkey.from_string(mint)) return f"Token supply: {supply}" @mcp.tool() async def request_airdrop(address: str, lamports: int) -> str: """Request an airdrop of lamports to a Pubkey. Args: address (str): Public key of recipient lamports (int): Amount of lamports to request Returns: str: Airdrop request result """ async with AsyncClient(rpc_url) as client: result = await client.request_airdrop(Pubkey.from_string(address), lamports) return f"Airdrop request: {result}" @mcp.tool() async def send_transaction(txn: bytes) -> str: """Send a transaction that has already been signed and serialized into the wire format. Args: txn (bytes): Signed transaction as bytes Returns: str: Transaction send result """ async with AsyncClient(rpc_url) as client: result = await client.send_raw_transaction(txn) return f"Transaction sent: {result}" @mcp.tool() async def validator_exit() -> str: """Request to have the validator exit. Returns: str: Validator exit request result """ async with AsyncClient(rpc_url) as client: result = await client.validator_exit() return f"Validator exit request: {result}"