XSpoonAi DAO Tooling
Automate DAO governance participation with SpoonOS agents.
Supported Platforms
| Platform | Type | Chains |
|---|---|---|
| Snapshot | Off-chain voting | All EVM |
| Tally | On-chain Governor | ETH, Polygon, Arbitrum, Optimism |
| Compound Governor | On-chain | ETH |
| Aragon | On-chain DAO | ETH, Polygon |
Snapshot Integration
API Configuration
SNAPSHOT_API = "https://hub.snapshot.org/graphql"
SNAPSHOT_SCORE_API = "https://score.snapshot.org"
Proposal Monitor
# scripts/snapshot_monitor.py
import aiohttp
from spoon_ai.tools.base import BaseTool
from pydantic import Field
from datetime import datetime
class SnapshotMonitorTool(BaseTool):
name: str = "snapshot_proposals"
description: str = "Monitor Snapshot governance proposals"
parameters: dict = Field(default={
"type": "object",
"properties": {
"space": {"type": "string", "description": "Snapshot space ID (e.g., 'uniswap.eth')"},
"state": {"type": "string", "enum": ["active", "pending", "closed", "all"], "default": "active"},
"limit": {"type": "integer", "default": 10}
},
"required": ["space"]
})
async def execute(self, space: str, state: str = "active", limit: int = 10) -> str:
query = """
query Proposals($space: String!, $state: String!, $limit: Int!) {
proposals(
first: $limit,
where: { space: $space, state: $state },
orderBy: "created",
orderDirection: desc
) {
id
title
body
choices
start
end
state
scores
scores_total
votes
author
space {
id
name
}
}
}
"""
variables = {
"space": space,
"state": state if state != "all" else None,
"limit": limit
}
async with aiohttp.ClientSession() as session:
async with session.post(
SNAPSHOT_API,
json={"query": query, "variables": variables}
) as resp:
if resp.status != 200:
return f"Error: {resp.status}"
data = await resp.json()
proposals = data.get("data", {}).get("proposals", [])
if not proposals:
return f"No {state} proposals found for {space}"
report = f"SNAPSHOT PROPOSALS: {space}\n{'='*50}\n\n"
for p in proposals:
end_time = datetime.fromtimestamp(p["end"])
time_left = end_time - datetime.now()
report += f"""
{p['title']}
ID: {p['id'][:20]}...
State: {p['state'].upper()}
Votes: {p['votes']:,}
"""
# Show choices with scores
if p['choices'] and p['scores']:
report += "Results:\n"
for i, choice in enumerate(p['choices']):
score = p['scores'][i] if i < len(p['scores']) else 0
total = p['scores_total'] or 1
pct = (score / total) * 100
report += f" - {choice}: {pct:.1f}%\n"
if p['state'] == 'active':
report += f"Ends: {end_time.strftime('%Y-%m-%d %H:%M')} ({time_left.days}d {time_left.seconds//3600}h left)\n"
report += "\n"
return report
Snapshot Voting
# scripts/snapshot_vote.py
import aiohttp
from eth_account import Account
from eth_account.messages import encode_defunct
import json
import time
class SnapshotVoteTool(BaseTool):
name: str = "snapshot_vote"
description: str = "Cast vote on Snapshot proposal"
parameters: dict = Field(default={
"type": "object",
"properties": {
"proposal_id": {"type": "string"},
"choice": {"type": "integer", "description": "Choice index (1-based)"},
"reason": {"type": "string", "default": ""}
},
"required": ["proposal_id", "choice"]
})
async def execute(self, proposal_id: str, choice: int, reason: str = "") -> str:
# Get proposal details first
proposal = await self._get_proposal(proposal_id)
if not proposal:
return f"Proposal {proposal_id} not found"
space = proposal["space"]["id"]
# Prepare vote message
account = Account.from_key(os.getenv("PRIVATE_KEY"))
vote_data = {
"space": space,
"proposal": proposal_id,
"type": "single-choice",
"choice": choice,
"reason": reason,
"app": "spoonos-agent",
"metadata": "{}"
}
# Sign the vote
message = json.dumps(vote_data, separators=(',', ':'))
signable = encode_defunct(text=message)
signature = account.sign_message(signable)
# Submit vote
payload = {
"address": account.address,
"sig": signature.signature.hex(),
"data": vote_data
}
async with aiohttp.ClientSession() as session:
async with session.post(
"https://seq.snapshot.org",
json=payload
) as resp:
if resp.status != 200:
error = await resp.text()
return f"Vote failed: {error}"
result = await resp.json()
choice_text = proposal["choices"][choice - 1]
return f"""
VOTE SUBMITTED
{'='*40}
Proposal: {proposal['title']}
Choice: {choice_text}
Voter: {account.address}
Receipt: {result.get('id', 'N/A')}
"""
async def _get_proposal(self, proposal_id: str) -> dict:
query = """
query Proposal($id: String!) {
proposal(id: $id) {
id
title
choices
space {
id
name
}
}
}
"""
async with aiohttp.ClientSession() as session:
async with session.post(
SNAPSHOT_API,
json={"query": query, "variables": {"id": proposal_id}}
) as resp:
data = await resp.json()
return data.get("data", {}).get("proposal")
Tally Integration
Tally API
TALLY_API = "https://api.tally.xyz/query"
class TallyMonitorTool(BaseTool):
name: str = "tally_proposals"
description: str = "Monitor on-chain governance via Tally"
parameters: dict = Field(default={
"type": "object",
"properties": {
"governor": {"type": "string", "description": "Governor contract or organization slug"},
"status": {"type": "string", "enum": ["active", "pending", "executed", "all"], "default": "active"}
},
"required": ["governor"]
})
async def execute(self, governor: str, status: str = "active") -> str:
query = """
query Proposals($governor: Address!, $status: ProposalStatusType) {
proposals(
chainId: "eip155:1"
governors: [$governor]
statuses: [$status]
first: 10
) {
nodes {
id
title
description
status
votes {
for
against
abstain
}
proposer {
address
name
}
voteStats {
percent
type
}
}
}
}
"""
headers = {
"Api-Key": os.getenv("TALLY_API_KEY"),
"Content-Type": "application/json"
}
async with aiohttp.ClientSession() as session:
async with session.post(
TALLY_API,
headers=headers,
json={"query": query, "variables": {"governor": governor, "status": status.upper()}}
) as resp:
if resp.status != 200:
return f"Error: {resp.status}"
data = await resp.json()
proposals = data.get("data", {}).get("proposals", {}).get("nodes", [])
report = f"TALLY GOVERNANCE: {governor}\n{'='*50}\n\n"
for p in proposals:
votes = p.get("votes", {})
report += f"""
{p['title']}
Status: {p['status']}
Proposer: {p['proposer']['name'] or p['proposer']['address'][:10]}...
Votes:
For: {int(votes.get('for', 0)):,}
Against: {int(votes.get('against', 0)):,}
Abstain: {int(votes.get('abstain', 0)):,}
"""
return report
On-Chain Voting
Governor Contract Integration
# scripts/governor_vote.py
from web3 import Web3
from spoon_ai.tools.base import BaseTool
from pydantic import Field
# Standard Governor ABI functions
GOVERNOR_ABI = [
{
"name": "castVote",
"inputs": [
{"name": "proposalId", "type": "uint256"},
{"name": "support", "type": "uint8"}
],
"outputs": [{"name": "", "type": "uint256"}],
"type": "function"
},
{
"name": "castVoteWithReason",
"inputs": [
{"name": "proposalId", "type": "uint256"},
{"name": "support", "type": "uint8"},
{"name": "reason", "type": "string"}
],
"outputs": [{"name": "", "type": "uint256"}],
"type": "function"
},
{
"name": "state",
"inputs": [{"name": "proposalId", "type": "uint256"}],
"outputs": [{"name": "", "type": "uint8"}],
"type": "function",
"stateMutability": "view"
}
]
# Vote support values
VOTE_AGAINST = 0
VOTE_FOR = 1
VOTE_ABSTAIN = 2
class GovernorVoteTool(BaseTool):
name: str = "governor_vote"
description: str = "Cast on-chain vote on Governor proposal"
parameters: dict = Field(default={
"type": "object",
"properties": {
"governor_address": {"type": "string"},
"proposal_id": {"type": "string"},
"support": {"type": "string", "enum": ["for", "against", "abstain"]},
"reason": {"type": "string", "default": ""},
"chain": {"type": "string", "default": "ethereum"}
},
"required": ["governor_address", "proposal_id", "support"]
})
async def execute(self, governor_address: str, proposal_id: str,
support: str, reason: str = "", chain: str = "ethereum") -> str:
w3 = Web3(Web3.HTTPProvider(RPC_URLS[chain]))
account = Account.from_key(os.getenv("PRIVATE_KEY"))
contract = w3.eth.contract(
address=Web3.to_checksum_address(governor_address),
abi=GOVERNOR_ABI
)
# Map support string to uint8
support_value = {
"for": VOTE_FOR,
"against": VOTE_AGAINST,
"abstain": VOTE_ABSTAIN
}[support]
# Check proposal state
state = contract.functions.state(int(proposal_id)).call()
if state != 1: # 1 = Active
states = ["Pending", "Active", "Canceled", "Defeated",
"Succeeded", "Queued", "Expired", "Executed"]
return f"Cannot vote: Proposal is {states[state]}"
# Build transaction
if reason:
tx_func = contract.functions.castVoteWithReason(
int(proposal_id),
support_value,
reason
)
else:
tx_func = contract.functions.castVote(
int(proposal_id),
support_value
)
tx = tx_func.build_transaction({
"from": account.address,
"nonce": w3.eth.get_transaction_count(account.address),
"gas": 200000,
"maxFeePerGas": w3.eth.gas_price * 2,
"maxPriorityFeePerGas": w3.to_wei(2, "gwei")
})
signed = account.sign_transaction(tx)
tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
if receipt.status == 1:
return f"""
ON-CHAIN VOTE SUBMITTED
{'='*40}
Governor: {governor_address}
Proposal: {proposal_id}
Vote: {support.upper()}
Tx: {tx_hash.hex()}
Gas Used: {receipt.gasUsed:,}
"""
else:
return f"Vote transaction failed: {tx_hash.hex()}"
Delegation Management
# scripts/delegation.py
class DelegationTool(BaseTool):
name: str = "manage_delegation"
description: str = "Delegate voting power to another address"
parameters: dict = Field(default={
"type": "object",
"properties": {
"token_address": {"type": "string", "description": "Governance token address"},
"delegate_to": {"type": "string", "description": "Address to delegate to"},
"chain": {"type": "string", "default": "ethereum"}
},
"required": ["token_address", "delegate_to"]
})
async def execute(self, token_address: str, delegate_to: str,
chain: str = "ethereum") -> str:
w3 = Web3(Web3.HTTPProvider(RPC_URLS[chain]))
account = Account.from_key(os.getenv("PRIVATE_KEY"))
# ERC20Votes delegate function
delegate_abi = [{
"name": "delegate",
"inputs": [{"name": "delegatee", "type": "address"}],
"outputs": [],
"type": "function"
}]
contract = w3.eth.contract(
address=Web3.to_checksum_address(token_address),
abi=delegate_abi
)
tx = contract.functions.delegate(
Web3.to_checksum_address(delegate_to)
).build_transaction({
"from": account.address,
"nonce": w3.eth.get_transaction_count(account.address),
"gas": 100000,
"maxFeePerGas": w3.eth.gas_price * 2,
"maxPriorityFeePerGas": w3.to_wei(2, "gwei")
})
signed = account.sign_transaction(tx)
tx_hash = w3.eth.send_raw_transaction(signed.raw_transaction)
receipt = w3.eth.wait_for_transaction_receipt(tx_hash)
if receipt.status == 1:
return f"""
DELEGATION SUCCESSFUL
{'='*40}
Token: {token_address}
Delegated to: {delegate_to}
Tx: {tx_hash.hex()}
"""
else:
return f"Delegation failed: {tx_hash.hex()}"
DAO Agent Example
# scripts/dao_agent.py
from spoon_ai.agents import SpoonReactMCP
from spoon_ai.chat import ChatBot
from spoon_ai.tools import ToolManager
class DAOGovernanceAgent(SpoonReactMCP):
name = "dao_governance_agent"
description = "Automated DAO governance participation"
system_prompt = """You are a DAO governance assistant.
CAPABILITIES:
- Monitor proposals on Snapshot and Tally
- Analyze proposal impact and voting patterns
- Execute votes (with user confirmation)
- Manage delegation settings
GOVERNANCE RULES:
- Always explain proposal implications before voting
- Require explicit confirmation for on-chain votes
- Track voting history and reasoning
- Consider quorum requirements
SUPPORTED PLATFORMS:
- Snapshot (off-chain, gasless)
- Tally (on-chain Governor contracts)
- Compound Governor
"""
def __init__(self):
super().__init__(
llm=ChatBot(model_name="gpt-4o"),
tools=ToolManager([
SnapshotMonitorTool(),
SnapshotVoteTool(),
TallyMonitorTool(),
GovernorVoteTool(),
DelegationTool()
]),
max_steps=10
)
Environment Variables
# API Keys
TALLY_API_KEY=...
# Wallet
PRIVATE_KEY=0x...
# RPC
ETHEREUM_RPC=https://eth.llamarpc.com
Best Practices
- Review Before Voting - Always analyze proposal impact
- Gas Optimization - Use Snapshot for non-critical votes
- Delegation Strategy - Delegate to active, aligned delegates
- Track Participation - Monitor voting power and history
- Quorum Awareness - Check quorum before voting