from flask import Flask, request, jsonify
from flask_cors import CORS
import os
from dotenv import load_dotenv
import requests
from cdp import Cdp, Wallet
import time
import traceback
# Initialize Flask app with CORS support for cross-origin requests
app = Flask(__name__)
CORS(app, resources={r"/api/*": {"origins": "*"}})
# Load environment variables from .env file
load_dotenv()
# Get required API keys from environment variables
CDP_API_KEY_NAME = os.getenv('CDP_API_KEY_NAME')
CDP_API_PRIVATE_KEY = os.getenv('CDP_API_PRIVATE_KEY')
PINATA_API_KEY = os.getenv('PINATA_API_KEY')
PINATA_SECRET_API_KEY = os.getenv('PINATA_SECRET_API_KEY')
# Configure CDP SDK with API credentials
if not CDP_API_KEY_NAME or not CDP_API_PRIVATE_KEY:
print("Missing CDP API credentials in environment variables")
raise EnvironmentError("CDP API credentials are required.")
Cdp.configure(CDP_API_KEY_NAME, CDP_API_PRIVATE_KEY)
# Track deployment progress for frontend updates
deployment_status = {
"step": "",
"status": ""
}
def update_deployment_status(step, status):
"""Helper function to update and log deployment status"""
deployment_status["step"] = step
deployment_status["status"] = status
print(f"Status Update: {step} - {status}")
def create_funded_wallet():
"""
Creates a new wallet on Base Sepolia testnet and funds it with test ETH.
Includes retry logic for faucet requests.
Returns:
Wallet: A CDP SDK wallet object with sufficient test ETH
"""
max_retries = 3
retry_delay = 5
try:
# Create new wallet
update_deployment_status("Creating new wallet on Base Sepolia...", "loading")
wallet = Wallet.create(network_id="base-sepolia")
print(f"Created wallet: {wallet.default_address.address_id}")
# Check if wallet already has sufficient funds
initial_balance = float(wallet.balance("eth"))
if initial_balance >= 0.01:
update_deployment_status("Wallet funded successfully.", "complete")
return wallet
# Request testnet ETH from faucet with retry logic
update_deployment_status("Requesting testnet ETH from faucet...", "loading")
for attempt in range(max_retries):
try:
wallet.faucet()
time.sleep(10) # Wait for faucet transaction to complete
current_balance = float(wallet.balance("eth"))
if current_balance >= 0.01:
update_deployment_status("Wallet funded successfully.", "complete")
return wallet
except Exception as e:
print(f"Faucet attempt {attempt + 1} failed: {str(e)}")
if attempt < max_retries - 1:
time.sleep(retry_delay)
else:
raise Exception("Failed to fund wallet after multiple attempts")
raise Exception("Failed to receive sufficient testnet ETH")
except Exception as e:
error_msg = str(e)
print(f"Error in create_funded_wallet: {error_msg}")
update_deployment_status("Creating new wallet on Base Sepolia...", "error")
raise Exception(error_msg)
@app.route('/api/deploy', methods=['POST'])
def deploy_contract():
"""
Main endpoint for deploying NFT contracts.
Handles both ERC721 (single NFTs) and ERC1155 (multi-token) deployments.
Expected POST data:
{
"type": "ERC721" or "ERC1155",
"name": "Contract Name", # For ERC721
"symbol": "SYMBOL", # For ERC721
"baseUri": "ipfs://<hash>/",
"tokenCount": number # For ERC1155
}
"""
try:
data = request.get_json()
print("\nReceived deployment request:", data)
# Create and fund a new wallet for deployment
wallet = create_funded_wallet()
if not wallet:
return jsonify({"success": False, "error": "Failed to create wallet with sufficient funds"}), 500
deployed_contract = None
max_retries = 3
try:
update_deployment_status("Deploying contract...", "loading")
for attempt in range(max_retries):
try:
if data['type'] == "ERC721":
# Deploy ERC721 NFT contract
base_uri = data['baseUri']
if not base_uri.endswith('/'):
base_uri += '/'
# Deploy contract with name, symbol, and base URI
deployed_contract = wallet.deploy_nft(
name=data['name'],
symbol=data['symbol'],
base_uri=base_uri
)
deployed_contract.wait()
print(f"ERC721 Contract deployed at: {deployed_contract.contract_address}")
# Mint initial NFT
mint_tx = wallet.invoke_contract(
contract_address=deployed_contract.contract_address,
method="mint",
args={"to": wallet.default_address.address_id}
)
mint_tx.wait()
print("Minted NFT with ID 1")
elif data['type'] == "ERC1155":
# Deploy ERC1155 Multi-token contract
base_uri = data['baseUri']
if not base_uri.endswith('/'):
base_uri += '/'
# Deploy contract with base URI
deployed_contract = wallet.deploy_multi_token(uri=base_uri)
deployed_contract.wait()
print(f"ERC1155 Contract deployed at: {deployed_contract.contract_address}")
# Mint specified number of tokens
for token_id in range(1, data.get('tokenCount', 1) + 1):
mint_tx = wallet.invoke_contract(
contract_address=deployed_contract.contract_address,
method="mint",
args={
"to": wallet.default_address.address_id,
"id": str(token_id),
"value": "1"
}
)
mint_tx.wait()
print(f"Minted token {token_id}")
break # Exit retry loop if successful
except Exception as e:
if attempt == max_retries - 1:
raise e
time.sleep(5)
continue
# Return successful deployment response
update_deployment_status("Minting tokens...", "complete")
return jsonify({
"success": True,
"contract_address": deployed_contract.contract_address,
"wallet_address": wallet.default_address.address_id,
"wallet_balance": str(wallet.balance("eth")),
"metadata_mapping": data.get('metadata', {})
})
except Exception as deploy_error:
print(f"Deployment error: {str(deploy_error)}")
traceback.print_exc()
update_deployment_status("Deploying contract...", "error")
return jsonify({
"success": False,
"error": f"Contract deployment failed: {str(deploy_error)}"
}), 500
except Exception as e:
print(f"Unexpected error: {str(e)}")
traceback.print_exc()
return jsonify({"success": False, "error": str(e)}), 500
# Run Flask app if executed directly
if __name__ == '__main__':
app.run(host="0.0.0.0", port=5328, debug=True)