my-project-public

repository

loading code, commits, and activity

repositories

loading repo index

my-project-public — gitlawb

#1	"""CDP (Coinbase Developer Platform) Client for Solana Account Management"""
#2
#3	import os
#4	import time
#5	import base64
#6	import asyncio
#7	from typing import Optional, List, Dict, Any
#8	from pathlib import Path
#9
#10	try:
#11	from cdp import CdpClient
#12	CDP_AVAILABLE = True
#13	except ImportError:
#14	CDP_AVAILABLE = False
#15	CdpClient = None
#16
#17	from solana.rpc.api import Client as SolanaClient
#18	from solana.rpc.types import TxOpts
#19	from solders.pubkey import Pubkey as PublicKey
#20	from solders.system_program import TransferParams, transfer
#21	from solders.message import Message
#22
#23
#24	class CDPSolanaClient:
#25	"""Client for CDP Solana account operations"""
#26
#27	def __init__(
#28	self,
#29	api_key_id: str,
#30	api_key_secret: str,
#31	wallet_secret: Optional[str] = None,
#32	rpc_url: str = "https://api.mainnet-beta.solana.com",
#33	network: str = "solana-mainnet",
#34	):
#35	"""
#36	Initialize CDP Solana client.
#37
#38	Args:
#39	api_key_id: CDP API Key ID
#40	api_key_secret: CDP API Key Secret (Ed25519 base64 or PEM EC key)
#41	wallet_secret: Optional wallet secret for signing transactions
#42	rpc_url: Solana RPC URL
#43	network: Network identifier (solana-mainnet or solana-devnet)
#44	"""
#45	if not CDP_AVAILABLE:
#46	raise ImportError(
#47	"cdp-sdk not installed. Run: pip install cdp-sdk"
#48	)
#49
#50	self.api_key_id = api_key_id
#51	self.api_key_secret = api_key_secret
#52	self.wallet_secret = wallet_secret
#53	self.rpc_url = rpc_url
#54	self.network = network
#55
#56	# Initialize CDP client
#57	self.cdp = CdpClient(
#58	api_key_id=api_key_id,
#59	api_key_secret=api_key_secret,
#60	wallet_secret=wallet_secret,
#61	)
#62
#63	# Initialize Solana RPC connection
#64	self.connection = SolanaClient(rpc_url)
#65
#66	# Cache for created accounts
#67	self._accounts_cache: Dict[str, Any] = {}
#68
#69	def _get_explorer_url(self, signature: str) -> str:
#70	"""Get Solana Explorer URL for a transaction."""
#71	if self.network == "solana-devnet":
#72	return f"https://explorer.solana.com/tx/{signature}?cluster=devnet"
#73	else:
#74	return f"https://explorer.solana.com/tx/{signature}"
#75
#76	async def create_account(self, name: Optional[str] = None) -> Dict[str, Any]:
#77	"""
#78	Create a new Solana account via CDP.
#79
#80	Args:
#81	name: Optional unique name for the account (2-36 chars, alphanumeric + hyphens)
#82
#83	Returns:
#84	Account info with address and metadata
#85	"""
#86	if name:
#87	account = await self.cdp.solana.create_account(name=name)
#88	else:
#89	account = await self.cdp.solana.create_account()
#90
#91	# Cache the account
#92	self._accounts_cache[account.address] = account
#93
#94	return {
#95	"address": account.address,
#96	"name": getattr(account, "name", None),
#97	"created_at": getattr(account, "created_at", None),
#98	}
#99
#100	async def get_account(self, address: str) -> Dict[str, Any]:
#101	"""
#102	Get a Solana account by address.
#103
#104	Args:
#105	address: Base58 encoded Solana address
#106
#107	Returns:
#108	Account info
#109	"""
#110	account = await self.cdp.solana.get_account(address)
#111	return {
#112	"address": account.address,
#113	"name": getattr(account, "name", None),
#114	"policies": getattr(account, "policies", []),
#115	"created_at": getattr(account, "created_at", None),
#116	"updated_at": getattr(account, "updated_at", None),
#117	}
#118
#119	async def list_accounts(self) -> List[Dict[str, Any]]:
#120	"""
#121	List all Solana accounts in the CDP project.
#122
#123	Returns:
#124	List of account info dicts
#125	"""
#126	result = await self.cdp.solana.list_accounts()
#127
#128	# CDP SDK returns (accounts_list, next_page_token) tuple
#129	# or a dict-like object with 'accounts' key
#130	if hasattr(result, 'accounts'):
#131	accounts = result.accounts
#132	elif isinstance(result, tuple) and len(result) >= 1:
#133	accounts = result[0]
#134	elif isinstance(result, dict) and 'accounts' in result:
#135	accounts = result['accounts']
#136	else:
#137	accounts = result
#138
#139	return [
#140	{
#141	"address": acc.address if hasattr(acc, 'address') else str(acc).split(': ')[-1],
#142	"name": getattr(acc, "name", None),
#143	}
#144	for acc in accounts
#145	]
#146
#147	async def request_faucet(self, address: str, token: str = "sol") -> Dict[str, Any]:
#148	"""
#149	Request funds from the Solana devnet faucet.
#150
#151	NOTE: Faucet only works on devnet. For mainnet, you must fund accounts manually.
#152
#153	Args:
#154	address: Solana address to fund
#155	token: Token to request ("sol" for native SOL)
#156
#157	Returns:
#158	Faucet response with transaction signature
#159	"""
#160	if self.network != "solana-devnet":
#161	raise ValueError("Faucet is only available on devnet. For mainnet, fund accounts manually.")
#162
#163	response = await self.cdp.solana.request_faucet(address, token=token)
#164	return {
#165	"transaction_signature": response.transaction_signature,
#166	"explorer_url": f"https://explorer.solana.com/tx/{response.transaction_signature}?cluster=devnet",
#167	}
#168
#169	async def get_balance(self, address: str) -> Dict[str, Any]:
#170	"""
#171	Get SOL balance for an address.
#172
#173	Args:
#174	address: Solana address
#175
#176	Returns:
#177	Balance info in SOL and lamports
#178	"""
#179	balance_resp = self.connection.get_balance(PublicKey.from_string(address))
#180	balance = balance_resp.value
#181
#182	return {
#183	"address": address,
#184	"balance_lamports": balance,
#185	"balance_sol": balance / 1e9,
#186	}
#187
#188	async def wait_for_balance(
#189	self,
#190	address: str,
#191	min_balance: int = 1,
#192	max_attempts: int = 30,
#193	) -> Dict[str, Any]:
#194	"""
#195	Wait for an account to be funded.
#196
#197	Args:
#198	address: Solana address
#199	min_balance: Minimum balance to wait for (in lamports)
#200	max_attempts: Maximum polling attempts
#201
#202	Returns:
#203	Final balance info
#204	"""
#205	balance = 0
#206	attempts = 0
#207
#208	while balance < min_balance and attempts < max_attempts:
#209	balance_resp = self.connection.get_balance(PublicKey.from_string(address))
#210	balance = balance_resp.value
#211	if balance < min_balance:
#212	await asyncio.sleep(1)
#213	attempts += 1
#214
#215	if balance < min_balance:
#216	raise TimeoutError(f"Account not funded after {max_attempts} attempts")
#217
#218	return {
#219	"address": address,
#220	"balance_lamports": balance,
#221	"balance_sol": balance / 1e9,
#222	"attempts": attempts,
#223	}
#224
#225	async def sign_transaction(self, address: str, transaction: str) -> Dict[str, Any]:
#226	"""
#227	Sign a Solana transaction using CDP.
#228
#229	Args:
#230	address: Signer address
#231	transaction: Base64 encoded transaction
#232
#233	Returns:
#234	Signed transaction response
#235	"""
#236	response = await self.cdp.solana.sign_transaction(
#237	address, transaction=transaction
#238	)
#239	return {
#240	"signed_transaction": response.signed_transaction,
#241	}
#242
#243	async def send_sol(
#244	self,
#245	from_address: str,
#246	to_address: str,
#247	lamports: int = 1000,
#248	) -> Dict[str, Any]:
#249	"""
#250	Send SOL from one account to another.
#251
#252	Args:
#253	from_address: Sender address (must be CDP-managed)
#254	to_address: Recipient address
#255	lamports: Amount in lamports (default: 1000 = 0.000001 SOL)
#256
#257	Returns:
#258	Transaction result with signature
#259	"""
#260	from_pubkey = PublicKey.from_string(from_address)
#261	to_pubkey = PublicKey.from_string(to_address)
#262
#263	# Get latest blockhash
#264	blockhash_resp = self.connection.get_latest_blockhash()
#265	blockhash = blockhash_resp.value.blockhash
#266
#267	# Create transfer instruction
#268	transfer_params = TransferParams(
#269	from_pubkey=from_pubkey,
#270	to_pubkey=to_pubkey,
#271	lamports=lamports,
#272	)
#273	transfer_instr = transfer(transfer_params)
#274
#275	# Build message
#276	message = Message.new_with_blockhash(
#277	[transfer_instr],
#278	from_pubkey,
#279	blockhash,
#280	)
#281
#282	# Create transaction envelope with signature space
#283	sig_count = bytes([1])
#284	empty_sig = bytes([0] * 64)
#285	message_bytes = bytes(message)
#286	tx_bytes = sig_count + empty_sig + message_bytes
#287
#288	# Encode to base64 for CDP API
#289	serialized_tx = base64.b64encode(tx_bytes).decode("utf-8")
#290
#291	# Sign with CDP
#292	signed_tx_response = await self.cdp.solana.sign_transaction(
#293	from_address,
#294	transaction=serialized_tx,
#295	)
#296
#297	# Decode and send
#298	decoded_signed_tx = base64.b64decode(signed_tx_response.signed_transaction)
#299
#300	tx_resp = self.connection.send_raw_transaction(
#301	decoded_signed_tx,
#302	opts=TxOpts(skip_preflight=False, preflight_commitment="processed"),
#303	)
#304	signature = tx_resp.value
#305
#306	# Wait for confirmation
#307	confirmation = self.connection.confirm_transaction(
#308	signature, commitment="processed"
#309	)
#310
#311	if hasattr(confirmation, "err") and confirmation.err:
#312	raise ValueError(f"Transaction failed: {confirmation.err}")
#313
#314	return {
#315	"signature": str(signature),
#316	"from_address": from_address,
#317	"to_address": to_address,
#318	"lamports": lamports,
#319	"sol_amount": lamports / 1e9,
#320	"explorer_url": self._get_explorer_url(str(signature)),
#321	}
#322
#323	async def sign_message(self, address: str, message: str) -> Dict[str, Any]:
#324	"""
#325	Sign a message using CDP.
#326
#327	Args:
#328	address: Signer address
#329	message: Message to sign
#330
#331	Returns:
#332	Signed message response
#333	"""
#334	encoded_message = base64.b64encode(message.encode()).decode("utf-8")
#335	response = await self.cdp.solana.sign_message(address, message=encoded_message)
#336	return {
#337	"signature": response.signature,
#338	"address": address,
#339	}
#340
#341	async def close(self):
#342	"""Close the CDP client connection."""
#343	await self.cdp.close()
#344
#345
#346	# Factory function for easy initialization
#347	def create_cdp_client(
#348	api_key_id: Optional[str] = None,
#349	api_key_secret: Optional[str] = None,
#350	wallet_secret: Optional[str] = None,
#351	rpc_url: str = "https://api.mainnet-beta.solana.com",
#352	network: str = "solana-mainnet",
#353	) -> Optional[CDPSolanaClient]:
#354	"""
#355	Create a CDP client from environment variables or parameters.
#356
#357	Args:
#358	api_key_id: CDP API Key ID (or CDP_API_KEY_ID env var)
#359	api_key_secret: CDP API Key Secret (or CDP_API_KEY_SECRET env var)
#360	wallet_secret: Optional wallet secret (or CDP_WALLET_SECRET env var)
#361	rpc_url: Solana RPC URL (defaults to mainnet)
#362	network: Network identifier - "solana-mainnet" or "solana-devnet"
#363
#364	Returns None if CDP SDK is not available or credentials are missing.
#365	"""
#366	if not CDP_AVAILABLE:
#367	return None
#368
#369	api_key_id = api_key_id or os.getenv("CDP_API_KEY_ID")
#370	api_key_secret = api_key_secret or os.getenv("CDP_API_KEY_SECRET")
#371	wallet_secret = wallet_secret or os.getenv("CDP_WALLET_SECRET")
#372
#373	if not api_key_id or not api_key_secret:
#374	return None
#375
#376	return CDPSolanaClient(
#377	api_key_id=api_key_id,
#378	api_key_secret=api_key_secret,
#379	wallet_secret=wallet_secret,
#380	rpc_url=rpc_url,
#381	network=network,
#382	)
#383

z6Mkq5mY3JWtxoxUobWcfNHm7AkRubgSWEZTkBVqZXJviFZ5/my-project-public