Building AI Agents for Blockchain Interaction with EVM MCP Protocol
April 1, 2025•ai
Introduction to EVM MCP Protocol
The EVM MCP (Ethereum Virtual Machine Message Call Protocol) is a standardized protocol for establishing connections between AI agents and EVM-compatible blockchain networks. Built on the foundation of the open-source evm-mcp-server, this protocol creates a seamless interface for AI systems to interact with smart contracts, tokens, and decentralized applications (dApps).
MCP serves as an abstraction layer that shields AI developers from many of the complexities involved in blockchain interactions, offering a more accessible pathway for creating autonomous agents that can deploy, interact with, and manage blockchain-based systems.
Key Components of the MCP Protocol
1. The MCP Server Architecture
The core of the MCP implementation is the server that manages connections between AI agents and blockchain networks. The server:
- Handles authentication and authorization
- Processes and routes message calls
- Manages transaction signing and submission
- Provides standardized error handling and status reporting
2. Request Schema
The MCP protocol uses a structured JSON format for all requests:
{
"chainId": 1,
"to": "0x742d35Cc6634C0532925a3b844Bc454e4438f44e",
"data": "0x...",
"value": "0",
"from": "0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045"
}
Key fields include:
chainId: The target blockchain network IDto: The recipient smart contract addressdata: The calldata payload (encoded function calls)value: The amount of native token to sendfrom: The sender address
3. The RPC Interface
The MCP server exposes a JSON-RPC interface that follows standard Ethereum patterns:
mcp_call: For read-only interactions with contractsmcp_sendTransaction: For state-changing operationsmcp_getTransactionReceipt: For retrieving execution resultsmcp_getTransactionByHash: For retrieving transaction details
Building an AI Agent for Blockchain Interaction
Prerequisites
Before building your blockchain-interacting AI agent, ensure you have:
- Access to an MCP server instance (self-hosted or via a provider)
- API credentials for the MCP server
- Understanding of AI systems and natural language processing
- Basic familiarity with blockchain concepts and Ethereum specifically
- Python development environment with necessary libraries
Step 1: Setting Up the Environment
First, set up your development environment and connect to the MCP server:
import os
import json
import requests
from dotenv import load_dotenv
import openai # or your preferred AI framework
# Load environment variables
load_dotenv()
# MCP server configuration
MCP_SERVER_URL = os.getenv('MCP_SERVER_URL')
MCP_API_KEY = os.getenv('MCP_API_KEY')
# Configure AI client
ai_client = openai.Client(api_key=os.getenv('OPENAI_API_KEY'))
# MCP server headers
headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {MCP_API_KEY}'
}
Step 2: Creating the MCP Client Interface
Implement a client interface for the MCP server:
class MCPClient:
def __init__(self, server_url, api_key):
self.server_url = server_url
self.headers = {
'Content-Type': 'application/json',
'Authorization': f'Bearer {api_key}'
}
self.request_id = 1
def _make_rpc_call(self, method, params):
"""Make a JSON-RPC call to the MCP server"""
payload = {
"jsonrpc": "2.0",
"method": method,
"params": params,
"id": self.request_id
}
self.request_id += 1
response = requests.post(
self.server_url,
headers=self.headers,
json=payload
)
result = response.json()
if "error" in result:
raise Exception(f"MCP Error: {result['error']['message']}")
return result["result"]
def call(self, chain_id, to, data, from_address=None, value="0"):
"""Make a read-only call to a smart contract"""
params = [{
"chainId": chain_id,
"to": to,
"data": data,
"value": value
}]
if from_address:
params[0]["from"] = from_address
return self._make_rpc_call("mcp_call", params)
def send_transaction(self, chain_id, to, data, from_address, value="0"):
"""Send a transaction to the blockchain"""
params = [{
"chainId": chain_id,
"to": to,
"data": data,
"value": value,
"from": from_address
}]
return self._make_rpc_call("mcp_sendTransaction", params)
def get_transaction_receipt(self, tx_hash):
"""Get the receipt for a transaction"""
return self._make_rpc_call("mcp_getTransactionReceipt", [tx_hash])
Step 3: Implementing the Intent Recognition System
Create a system that translates natural language instructions into blockchain operations:
class IntentRecognizer:
def __init__(self, ai_client, contract_abis=None):
self.ai_client = ai_client
self.contract_abis = contract_abis or {}
async def recognize_intent(self, user_input):
"""Process natural language to determine blockchain intent"""
response = await self.ai_client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "Translate the following request into a structured blockchain operation."},
{"role": "user", "content": user_input}
],
response_format={"type": "json_object"}
)
structured_intent = json.loads(response.choices[0].message.content)
return self._prepare_mcp_request(structured_intent)
def _prepare_mcp_request(self, intent):
"""Convert structured intent into MCP request format"""
operation_type = intent.get("operation_type")
if operation_type == "token_transfer":
return self._create_token_transfer_request(
intent.get("chain_id", 1),
intent.get("token_address"),
intent.get("recipient"),
intent.get("amount"),
intent.get("from_address")
)
elif operation_type == "contract_call":
return self._create_contract_call_request(
intent.get("chain_id", 1),
intent.get("contract_address"),
intent.get("function_name"),
intent.get("parameters", []),
intent.get("from_address"),
intent.get("value", "0")
)
else:
raise ValueError(f"Unsupported operation type: {operation_type}")
def _create_token_transfer_request(self, chain_id, token_address, recipient, amount, from_address):
"""Create ERC-20 token transfer request"""
# Get ABI for ERC-20 tokens
from web3 import Web3
import json
# Standard ERC-20 transfer function selector
# This is the keccak256 hash of "transfer(address,uint256)" signature
transfer_selector = "0xa9059cbb"
# Encode the parameters (recipient address and amount)
# First, pad the address to 32 bytes
padded_address = recipient.replace("0x", "").rjust(64, "0")
# Convert amount to wei and pad to 32 bytes
amount_wei = Web3.to_wei(amount, "ether")
padded_amount = hex(amount_wei)[2:].rjust(64, "0")
# Combine the function selector and encoded parameters
data = transfer_selector + padded_address + padded_amount
return {
"chainId": chain_id,
"to": token_address,
"data": "0x" + data,
"value": "0",
"from": from_address
}
def _create_contract_call_request(self, chain_id, contract_address, function_name, parameters, from_address, value):
"""Create contract call request with encoded data"""
# Here we would need the ABI for the specific contract
# For demonstration, we'll assume the ABI is available in self.contract_abis
abi = self.contract_abis.get(contract_address)
if not abi:
raise ValueError(f"ABI not found for contract {contract_address}")
from web3 import Web3
# Create contract instance
w3 = Web3()
contract = w3.eth.contract(address=contract_address, abi=abi)
# Get the function object
contract_function = getattr(contract.functions, function_name)
# Build the transaction data
tx_data = contract_function(*parameters).build_transaction({
"chainId": chain_id,
"gas": 2000000,
"maxFeePerGas": 2000000000,
"maxPriorityFeePerGas": 1000000000,
"nonce": 0 # This will be ignored in the request
})
return {
"chainId": chain_id,
"to": contract_address,
"data": tx_data["data"],
"value": value,
"from": from_address
}
Step 4: Building the Blockchain AI Agent
Now, integrate all components into a cohesive agent:
class BlockchainAIAgent:
def __init__(self, mcp_server_url, mcp_api_key, ai_api_key, wallet_address):
# Initialize MCP client
self.mcp_client = MCPClient(mcp_server_url, mcp_api_key)
# Initialize AI client
self.ai_client = openai.Client(api_key=ai_api_key)
# Initialize intent recognizer
self.intent_recognizer = IntentRecognizer(self.ai_client)
# Set wallet address
self.wallet_address = wallet_address
# Initialize memory for conversation context
self.memory = []
async def process_instruction(self, user_instruction):
"""Process a natural language instruction from the user"""
# Add to agent memory
self.memory.append({"role": "user", "content": user_instruction})
# Recognize intent and convert to MCP request
try:
mcp_request = await self.intent_recognizer.recognize_intent(user_instruction)
# Ensure from address is set
if "from" not in mcp_request or not mcp_request["from"]:
mcp_request["from"] = self.wallet_address
# Determine if this is a read or write operation
is_write_operation = self._is_state_changing(mcp_request)
if is_write_operation:
# Send transaction
tx_hash = self.mcp_client.send_transaction(
mcp_request["chainId"],
mcp_request["to"],
mcp_request["data"],
mcp_request["from"],
mcp_request.get("value", "0")
)
# Wait for receipt
receipt = self._wait_for_receipt(tx_hash)
result = {"tx_hash": tx_hash, "receipt": receipt}
else:
# Make read-only call
result = self.mcp_client.call(
mcp_request["chainId"],
mcp_request["to"],
mcp_request["data"],
mcp_request.get("from"),
mcp_request.get("value", "0")
)
# Generate response using AI
response = await self._generate_response(user_instruction, mcp_request, result)
# Add to agent memory
self.memory.append({"role": "assistant", "content": response})
return {
"response": response,
"operation_result": result
}
except Exception as e:
error_response = f"I encountered an error processing your request: {str(e)}"
self.memory.append({"role": "assistant", "content": error_response})
return {"response": error_response, "error": str(e)}
def _is_state_changing(self, request):
"""Determine if a request will change blockchain state"""
# This is a simplified check
# In practice, you would need to analyze the function signature
# If there's value being transferred, it's a state change
if request.get("value") and request["value"] != "0":
return True
# If there's no data, it's a simple ETH transfer (state changing)
if not request.get("data") or request["data"] == "0x":
return True
# Otherwise, analyze the function selector (first 4 bytes of data)
# This would require knowledge of the contract's ABI
# For simplicity, we'll assume any data payload means a state change
return True
def _wait_for_receipt(self, tx_hash, max_attempts=30):
"""Wait for a transaction receipt with timeout"""
for i in range(max_attempts):
try:
receipt = self.mcp_client.get_transaction_receipt(tx_hash)
if receipt:
return receipt
except Exception:
pass
# Wait before retrying
import time
time.sleep(2)
raise TimeoutError(f"Transaction {tx_hash} was not mined within the timeout period")
async def _generate_response(self, instruction, request, result):
"""Generate a human-friendly response about the operation"""
context = json.dumps({
"instruction": instruction,
"request": request,
"result": result
})
response = await self.ai_client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": f"You are a blockchain AI assistant. Explain the results of the following blockchain operation in simple terms:\n{context}"},
*self.memory[-5:] # Include recent conversation history
]
)
return response.choices[0].message.content
Practical Applications
1. Automated DeFi Portfolio Management
async def auto_rebalance_portfolio(self, target_allocations, chain_id=1):
"""
Automatically rebalance a DeFi portfolio based on target allocations
Args:
target_allocations: Dict mapping token addresses to target percentages
chain_id: The blockchain network to operate on
"""
# Get current token balances
current_balances = await self._get_token_balances(target_allocations.keys(), chain_id)
# Calculate total portfolio value in a stable reference (e.g., USD)
portfolio_value, token_values = await self._calculate_portfolio_value(current_balances, chain_id)
# Determine required trades to achieve target allocation
required_trades = []
for token_address, target_pct in target_allocations.items():
current_value = token_values.get(token_address, 0)
target_value = portfolio_value * (target_pct / 100)
# Calculate difference
value_diff = target_value - current_value
# If difference is significant (e.g., >1% of portfolio), add to trades
if abs(value_diff) > (portfolio_value * 0.01):
required_trades.append({
"token": token_address,
"direction": "buy" if value_diff > 0 else "sell",
"value_diff": abs(value_diff)
})
# Execute trades via a DEX (e.g., Uniswap)
for trade in required_trades:
if trade["direction"] == "buy":
await self._execute_token_swap(
"0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", # USDC as example
trade["token"],
trade["value_diff"],
chain_id
)
else:
await self._execute_token_swap(
trade["token"],
"0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48", # USDC as example
trade["value_diff"],
chain_id
)
return "Portfolio rebalanced successfully"
2. Autonomous DAO Governance Participation
async def analyze_and_vote_on_proposals(self, dao_address, chain_id=1):
"""
Analyze open proposals for a DAO and vote according to predefined strategies
Args:
dao_address: The address of the DAO governance contract
chain_id: The blockchain network to operate on
"""
# Get active proposals
active_proposals = await self._get_active_proposals(dao_address, chain_id)
for proposal in active_proposals:
# Check if already voted
has_voted = await self._check_if_voted(dao_address, proposal["id"], chain_id)
if has_voted:
continue
# Analyze proposal using AI
proposal_details = await self._get_proposal_details(dao_address, proposal["id"], chain_id)
analysis = await self.ai_client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content":
"You are a DAO governance analyst. Evaluate the following proposal and recommend a vote based on the established strategy." +
"The strategy prioritizes: 1) Protocol security, 2) Long-term value accrual, 3) Decentralization"},
{"role": "user", "content": json.dumps(proposal_details)}
],
response_format={"type": "json_object"}
)
vote_decision = json.loads(analysis.choices[0].message.content)
# Vote on the proposal
await self._cast_vote(
dao_address,
proposal["id"],
vote_decision["vote"], # "for", "against", or "abstain"
chain_id
)
return "Governance participation completed"
3. Automated NFT Trading
async def monitor_and_trade_nfts(self, collection_addresses, budget_eth, chain_id=1):
"""
Monitor NFT collections and automatically purchase undervalued assets
Args:
collection_addresses: List of NFT collection addresses to monitor
budget_eth: Maximum budget in ETH
chain_id: The blockchain network to operate on
"""
for collection in collection_addresses:
# Get collection floor price
floor_price = await self._get_collection_floor_price(collection, chain_id)
# Get recent sales
recent_sales = await self._get_recent_sales(collection, chain_id)
# Get listed NFTs
listed_nfts = await self._get_listed_nfts(collection, chain_id)
# Analyze each listed NFT for potential value
for nft in listed_nfts:
if float(nft["price"]) > float(budget_eth):
continue
# Get NFT metadata and traits
metadata = await self._get_nft_metadata(collection, nft["token_id"], chain_id)
# Compare with collection trait floor prices
trait_analysis = await self._analyze_nft_traits(collection, metadata["attributes"], chain_id)
# Use AI to determine if this is a good purchase
recommendation = await self.ai_client.chat.completions.create(
model="gpt-4",
messages=[
{"role": "system", "content": "You are an NFT trading assistant. Evaluate this NFT for potential purchase."},
{"role": "user", "content": json.dumps({
"collection": collection,
"token_id": nft["token_id"],
"price": nft["price"],
"floor_price": floor_price,
"metadata": metadata,
"trait_analysis": trait_analysis,
"recent_sales": recent_sales
})}
],
response_format={"type": "json_object"}
)
decision = json.loads(recommendation.choices[0].message.content)
if decision["should_buy"]:
# Execute purchase
await self._purchase_nft(collection, nft["token_id"], nft["price"], chain_id)
budget_eth = float(budget_eth) - float(nft["price"])
if budget_eth <= 0:
break
return "NFT trading session completed"
Advanced Integration Patterns
1. Multi-chain Operations
The MCP protocol supports operations across multiple EVM-compatible chains:
async def bridge_assets(self, source_chain_id, target_chain_id, token_address, amount):
"""
Bridge assets from one chain to another
Args:
source_chain_id: Source chain ID
target_chain_id: Target chain ID
token_address: Token to bridge
amount: Amount to bridge
"""
# Step 1: Approve the bridge contract on source chain
bridge_contract = self._get_bridge_address(source_chain_id, target_chain_id)
approve_request = {
"chainId": source_chain_id,
"to": token_address,
"data": self._encode_approve_call(bridge_contract, amount),
"from": self.wallet_address
}
approve_tx = self.mcp_client.send_transaction(
approve_request["chainId"],
approve_request["to"],
approve_request["data"],
approve_request["from"]
)
self._wait_for_receipt(approve_tx)
# Step 2: Execute bridge transaction
bridge_request = {
"chainId": source_chain_id,
"to": bridge_contract,
"data": self._encode_bridge_call(token_address, target_chain_id, amount),
"from": self.wallet_address
}
bridge_tx = self.mcp_client.send_transaction(
bridge_request["chainId"],
bridge_request["to"],
bridge_request["data"],
bridge_request["from"]
)
receipt = self._wait_for_receipt(bridge_tx)
# Step 3: Wait for confirmation on target chain
status = await self._monitor_bridge_status(source_chain_id, target_chain_id, receipt["logs"])
return status
2. Integration with AI-Powered Oracles
async def get_ai_oracle_prediction(self, query, oracle_address, chain_id=1):
"""
Query an on-chain AI oracle for predictions
Args:
query: The query string to send to the oracle
oracle_address: The oracle contract address
chain_id: The blockchain network where the oracle is deployed
"""
# Step 1: Generate a commitment hash for the query
import hashlib
query_hash = hashlib.sha256(query.encode()).hexdigest()
# Step 2: Format oracle request
oracle_request = {
"chainId": chain_id,
"to": oracle_address,
"data": self._encode_oracle_request(query_hash),
"value": "1000000000000000", # 0.001 ETH as fee
"from": self.wallet_address
}
# Step 3: Send request to oracle
request_tx = self.mcp_client.send_transaction(
oracle_request["chainId"],
oracle_request["to"],
oracle_request["data"],
oracle_request["from"],
oracle_request["value"]
)
receipt = self._wait_for_receipt(request_tx)
# Step 4: Extract request ID from logs
request_id = self._extract_request_id_from_logs(receipt["logs"])
# Step 5: Wait for oracle to respond
for i in range(30):
# Check if prediction is ready
check_request = {
"chainId": chain_id,
"to": oracle_address,
"data": self._encode_check_prediction(request_id),
"from": self.wallet_address
}
result = self.mcp_client.call(
check_request["chainId"],
check_request["to"],
check_request["data"],
check_request["from"]
)
if result != "0x":
# Decode the prediction result
prediction = self._decode_prediction_result(result)
return prediction
# Wait before checking again
import time
time.sleep(10)
raise TimeoutError("Oracle did not respond within the timeout period")
Best Practices and Security Considerations
1. Implementing Security Safeguards
class SecurityManager:
def __init__(self, wallet_address):
self.wallet_address = wallet_address
self.transaction_limits = {
"default": {
"daily": "1000000000000000000", # 1 ETH
"per_tx": "500000000000000000" # 0.5 ETH
},
"contracts": {} # Whitelist specific contracts
}
self.whitelist = {} # Approved addresses by function
self.blacklist = [] # Blocked addresses
def validate_transaction(self, tx_request):
"""Validate a transaction against security policies"""
# Check for blacklisted addresses
if tx_request["to"].lower() in [addr.lower() for addr in self.blacklist]:
raise SecurityException("Recipient address is blacklisted")
# Check transaction value limits
if "value" in tx_request and tx_request["value"] != "0":
value = int(tx_request["value"])
per_tx_limit = self.transaction_limits["default"]["per_tx"]
# Check for contract-specific limits
if tx_request["to"].lower() in self.transaction_limits["contracts"]:
per_tx_limit = self.transaction_limits["contracts"][tx_request["to"].lower()]["per_tx"]
if value > int(per_tx_limit):
raise SecurityException(f"Transaction value exceeds per-transaction limit of {per_tx_limit}")
# For contract interactions, validate function signature and parameters
if "data" in tx_request and tx_request["data"] != "0x":
function_sig = tx_request["data"][:10] # First 4 bytes + 0x prefix
# If we have a whitelist for this contract and function
if tx_request["to"].lower() in self.whitelist and function_sig in self.whitelist[tx_request["to"].lower()]:
# Validate parameters if needed
pass
return True
2. Implementing Rate Limiting and Transaction Monitoring
class TransactionMonitor:
def __init__(self):
self.tx_history = []
self.daily_totals = {}
def record_transaction(self, tx_request, tx_hash, receipt=None):
"""Record a transaction in the monitoring system"""
import time
# Create transaction record
tx_record = {
"timestamp": int(time.time()),
"request": tx_request,
"tx_hash": tx_hash,
"receipt": receipt,
"date": time.strftime("%Y-%m-%d")
}
# Add to history
self.tx_history.append(tx_record)
# Update daily totals
date = tx_record["date"]
if date not in self.daily_totals:
self.daily_totals[date] = 0
if "value" in tx_request and tx_request["value"] != "0":
self.daily_totals[date] += int(tx_request["value"])
return tx_record
def check_daily_limit(self, tx_request, daily_limit):
"""Check if a transaction would exceed the daily limit"""
import time
date = time.strftime("%Y-%m-%d")
current_total = self.daily_totals.get(date, 0)
if "value" in tx_request and tx_request["value"] != "0":
new_total = current_total + int(tx_request["value"])
if new_total > int(daily_limit):
return False
return True
Conclusion
The EVM MCP protocol provides a standardized interface for AI agents to interact with blockchain networks, abstracting away many of the complexities involved in direct blockchain interaction. By leveraging the open-source evm-mcp-server, developers can build sophisticated AI agents that can autonomously interact with smart contracts, manage assets, and participate in decentralized systems.
As blockchain technology and AI continue to evolve, the intersection of these fields offers exciting possibilities for autonomous financial systems, decentralized governance, and intelligent digital asset management. The MCP protocol represents an important step toward making these possibilities more accessible to developers and users alike.