main.py

main.py

# LangChain Blockchain Analysis MVP with Claude in Python
# This application analyzes blockchain transactions and provides natural language insights using LangChain with Claude

# Required dependencies
from langchain_anthropic import ChatAnthropic
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
from langchain.tools import Tool
from langchain.agents import initialize_agent, AgentType
from web3 import Web3
from dotenv import load_dotenv
from flask import Flask, request, jsonify
import os
import json

# Load environment variables
load_dotenv()

# Initialize Flask app
app = Flask(__name__)

# Initialize Ethereum provider
web3 = Web3(Web3.HTTPProvider(os.getenv('ETHEREUM_RPC_URL')))

# Initialize Claude model
model = ChatAnthropic(
    api_key=os.getenv('ANTHROPIC_API_KEY'),
    model_name="claude-3-7-sonnet-20250219",  # Use the latest Claude model
    temperature=0
)

# Helper functions for blockchain interactions
def get_transaction_details(tx_hash):
    try:
        tx_hash = tx_hash.strip()
        if not tx_hash.startswith('0x'):
            tx_hash = '0x' + tx_hash
            
        tx = web3.eth.get_transaction(tx_hash)
        receipt = web3.eth.get_transaction_receipt(tx_hash)
        
        return {
            'hash': tx_hash,
            'from': tx['from'],
            'to': tx['to'],
            'value': web3.from_wei(tx['value'], 'ether'),
            'gasPrice': web3.from_wei(tx['gasPrice'], 'gwei'),
            'confirmations': web3.eth.block_number - tx['blockNumber'] if tx['blockNumber'] else 0,
            'status': 'Success' if receipt['status'] == 1 else 'Failed' if receipt else 'Pending',
            'blockNumber': tx['blockNumber']
        }
    except Exception as e:
        return f"Error fetching transaction: {str(e)}"

def get_wallet_balance(address):
    try:
        address = address.strip()
        if not address.startswith('0x'):
            address = '0x' + address
            
        balance = web3.eth.get_balance(address)
        return f"The wallet {address} has {web3.from_wei(balance, 'ether')} ETH"
    except Exception as e:
        return f"Error fetching wallet balance: {str(e)}"

def analyze_transactions(address):
    try:
        address = address.strip()
        if not address.startswith('0x'):
            address = '0x' + address
            
        # Get current block number
        current_block = web3.eth.block_number
        
        # Look at the last 10 blocks for transactions
        transactions = []
        for i in range(10):
            block = web3.eth.get_block(current_block - i, full_transactions=True)
            address_txs = [tx for tx in block.transactions if 
                          tx['from'] == address or tx['to'] == address]
            transactions.extend(address_txs)
        
        return {
            'address': address,
            'transactionCount': len(transactions),
            'sent': len([tx for tx in transactions if tx['from'] == address]),
            'received': len([tx for tx in transactions if tx['to'] == address]),
            'transactions': [{
                'hash': tx['hash'].hex(),
                'from': tx['from'],
                'to': tx['to'],
                'value': web3.from_wei(tx['value'], 'ether')
            } for tx in transactions[:5]]
        }
    except Exception as e:
        return f"Error analyzing transactions: {str(e)}"

# Create tools for the agent
tools = [
    Tool(
        name="transaction_details",
        func=lambda tx_hash: json.dumps(get_transaction_details(tx_hash)),
        description="Get detailed information about a blockchain transaction. Input should be a transaction hash."
    ),
    Tool(
        name="wallet_balance",
        func=get_wallet_balance,
        description="Get the current balance of an Ethereum wallet address. Input should be a wallet address."
    ),
    Tool(
        name="transaction_analysis",
        func=lambda address: json.dumps(analyze_transactions(address)),
        description="Analyze recent transactions for a given address. Input should be a wallet address."
    )
]

# Initialize agent with tools
def initialize_blockchain_agent():
    agent = initialize_agent(
        tools,
        model,
        agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
        verbose=True
    )
    return agent

# API routes
@app.route('/api/blockchain/analyze', methods=['POST'])
def blockchain_analyze():
    try:
        data = request.json
        query = data.get('query')
        
        if not query:
            return jsonify({"error": "Query is required"}), 400
        
        agent = initialize_blockchain_agent()
        result = agent.run(query)
        
        return jsonify({"result": result})
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/transaction/<tx_hash>', methods=['GET'])
def transaction_details(tx_hash):
    try:
        tx_details = get_transaction_details(tx_hash)
        return jsonify(tx_details)
    except Exception as e:
        return jsonify({"error": str(e)}), 500

@app.route('/api/wallet/<address>', methods=['GET'])
def wallet_details(address):
    try:
        balance = web3.eth.get_balance(address)
        return jsonify({
            "address": address,
            "balance": web3.from_wei(balance, 'ether')
        })
    except Exception as e:
        return jsonify({"error": str(e)}), 500

# Main execution
if __name__ == '__main__':
    port = int(os.getenv('PORT', 3000))
    app.run(host='0.0.0.0', port=port, debug=True)