MSG Chain AI Agent CW20 代币发行与管理指南
链 ID:
msg-chain-1| 共识: DAR | 虚拟机: CosmWasm (WasmVM)
状态: 规划文档 — 主网裁决为 No-Go，所有数据均为主网预演
Gas: 1,000,000,000 attoMSG/gas
Gas 分配: 40% 验证者 / 30% 开发者 / 20% 燃烧 / 10% 基金会金库
地址格式: SHA3-512(前40位) + SHA-256 校验 | 签名: Dilithium-5 (公钥2592字节 / 私钥4864字节 / 签名4595字节)
AI Agent安全边界: 永不自主创建新合约，永不自主调整 Gas 参数，永不自主铸造/销毁代币
1. 概述
1.1 为什么 AI Agent 需要自己的代币
AI Agent 在 msg-chain-1 上运行，拥有独立的身份（DID）、存储和执行环境。发行自有代币赋予 Agent 以下能力：
- 经济自主：Agent 通过代币获得独立的经济基础，不再依赖外部资助
- 用户激励：代币用于奖励与 Agent 交互的用户，驱动网络效应
- 治理权力：代币持有者参与 Agent 的发展方向决策
- 资源定价：Agent 提供的服务通过代币定价和支付
- 社区建设：围绕 Agent 形成代币持有者社区
1.2 CW20 标准概述
CW20 是 CosmWasm 生态中的代币标准，类似以太坊的 ERC20。MSG Chain 原生支持 CW20，提供以下核心功能：
Transfer/TransferFrom：代币转账Balance：余额查询Approve/Allowance：授权机制Mint/Burn：增发与销毁Send/SendFrom：带 Hook 的转账
CW20 同时支持元数据扩展（名称、符号、精度）和快照扩展。
1.3 代币类型与适用场景
| 代币类型 | 用途 | 典型场景 |
|---|---|---|
| 实用代币 | 支付 Agent 服务费用 | 每次查询消耗 1 AGENT |
| 治理代币 | 社区投票决策 | 提案投票、参数调整 |
| 奖励代币 | 激励用户贡献 | 数据标注、内容创作 |
| 社交代币 | 社区身份与归属 | 持有门槛、专属频道 |
| 收益代币 | 分红与收益分配 | Agent 收入分成 |
1.4 本指南目标读者
- 在 MSG Chain 上部署 AI Agent 的开发者
- 希望为 Agent 引入代币经济的产品经理
- 研究 AI + 区块链代币模型的研究者
1.5 前置要求
- 熟悉 Rust 基础语法
- 了解 CosmWasm 智能合约开发
- 安装有
msgcli命令行工具 - 拥有 MSG Chain 钱包地址（msg 前缀）
1.6 技术栈概览
智能合约语言: Rust + CosmWasm
客户端脚本: Python 3.8+
链交互工具: msgcli
钱包格式: msg1...
合约部署: cosmwasm-opt + msgcli
1.7 目录说明
- 第2章：CW20 代币合约的完整实现与部署
- 第3章：代币分发策略，包括空投和 vesting
- 第4章：质押与奖励机制，提升代币效用
- 第5章：流动性池创建与管理
- 第6章：代币门控实现 Agent 的付费访问
- 第7章：治理代币与 DAO 集成
- 第8章：完整示例脚本
2. CW20 代币合约
2.1 合约架构概述
CW20 代币合约分为两层：
- 核心层：
cw20-base提供标准实现 - 扩展层：Agent 专属逻辑（身份绑定、权限控制）
合约文件结构：
agent-token/
├── Cargo.toml
├── src/
│ ├── contract.rs # 合约入口
│ ├── msg.rs # 消息定义
│ ├── state.rs # 状态管理
│ ├── execute.rs # 执行逻辑
│ └── query.rs # 查询逻辑
└── examples/
└── deploy.py # 部署脚本
2.2 Cargo.toml 依赖配置
[package]
name = "agent-token"
version = "0.1.0"
edition = "2021"
[lib]
crate-type = ["cdylib", "rlib"]
[features]
default = ["library"]
library = []
[dependencies]
cosmwasm-std = "2.0"
cosmwasm-storage = "2.0"
cw20 = "2.0"
cw20-base = "2.0"
cw-storage-plus = "2.0"
cw-utils = "2.0"
schemars = "0.8"
serde = { version = "1.0", features = ["derive"] }
thiserror = "1.0"
[profile.release]
opt-level = "z"
lto = true
codegen-units = 1
panic = "abort"
2.3 消息定义 (msg.rs)
use cosmwasm_std::Uint128;
use cw20::{Cw20Coin, MinterResponse};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
/// 实例化消息：Agent 代币的创建参数
#[derive(Serialize, Deserialize, JsonSchema, Debug, Clone)]
pub struct InstantiateMsg {
pub name: String,
pub symbol: String,
pub decimals: u8,
pub initial_balances: Vec<Cw20Coin>,
pub mint: Option<MinterResponse>,
/// Agent 的去中心化身份标识
pub agent_did: String,
/// 代币类型：utility | governance | reward | social
pub token_type: String,
/// Agent 的管理员地址
pub admin: String,
}
/// 执行消息：Agent 专属操作
#[derive(Serialize, Deserialize, JsonSchema, Debug, Clone)]
#[serde(rename_all = "snake_case")]
pub enum ExecuteMsg {
/// 标准 CW20 操作
Transfer { recipient: String, amount: Uint128 },
TransferFrom { owner: String, recipient: String, amount: Uint128 },
Approve { spender: String, amount: Uint128 },
Mint { recipient: String, amount: Uint128 },
Burn { amount: Uint128 },
Send { contract: String, amount: Uint128, msg: Binary },
/// Agent 专属操作
UpdateMetadata { new_did: Option<String>, new_admin: Option<String> },
Pause {},
Unpause {},
/// 批量操作
BatchTransfer { recipients: Vec<String>, amounts: Vec<Uint128> },
}
/// 查询消息
#[derive(Serialize, Deserialize, JsonSchema, Debug, Clone)]
#[serde(rename_all = "snake_case")]
pub enum QueryMsg {
/// 标准 CW20 查询
Balance { address: String },
TokenInfo {},
Minter {},
Allowance { owner: String, spender: String },
AllAllowances { owner: String, start_after: Option<String>, limit: Option<u32> },
/// Agent 专属查询
AgentMetadata {},
TokenType {},
IsPaused {},
}
/// 合约迁移消息
#[derive(Serialize, Deserialize, JsonSchema, Debug, Clone)]
pub struct MigrateMsg {
pub new_contract_version: String,
}
2.4 状态管理 (state.rs)
use cosmwasm_std::Addr;
use cw_storage_plus::{Item, Map};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
/// Agent 代币元数据
#[derive(Serialize, Deserialize, JsonSchema, Debug, Clone)]
pub struct TokenMetadata {
/// Agent DID
pub agent_did: String,
/// 创建时的区块高度
pub created_at: u64,
/// 合约管理员
pub admin: Addr,
/// 代币类型
pub token_type: String,
/// 是否暂停
pub paused: bool,
/// 总供应量上限（None 表示无上限）
pub cap: Option<Uint128>,
}
/// 存储键定义
pub const TOKEN_METADATA: Item<TokenMetadata> = Item::new("agent_metadata");
pub const AGENT_DID_LOOKUP: Map<&str, Addr> = Map::new("agent_did");
pub const PAUSED: Item<bool> = Item::new("paused");
/// 质押相关状态
pub const STAKED_BALANCES: Map<&Addr, Uint128> = Map::new("staked");
pub const REWARD_DEBTS: Map<&Addr, Uint128> = Map::new("rewards");
pub const TOTAL_STAKED: Item<Uint128> = Item::new("total_staked");
pub const REWARD_RATE: Item<Uint128> = Item::new("reward_rate");
pub const REWARD_INDEX: Item<Uint128> = Item::new("reward_index");
2.5 合约入口 (contract.rs)
use cosmwasm_std::{
entry_point, to_binary, Addr, Binary, Deps, DepsMut, Env,
MessageInfo, Response, StdError, StdResult, Uint128,
};
use cw20_base::contract::{
execute_transfer, execute_approve, execute_transfer_from,
execute_burn, execute_burn_from, execute_mint,
execute_send, execute_send_from, execute_update_marketing,
execute_upload_logo, query_balance, query_token_info,
query_minter, query_allowance, query_all_allowances,
query_marketing_info, query_download_logo,
};
use crate::msg::{ExecuteMsg, InstantiateMsg, MigrateMsg, QueryMsg};
use crate::state::{TokenMetadata, TOKEN_METADATA, PAUSED};
#[entry_point]
pub fn instantiate(
deps: DepsMut,
env: Env,
info: MessageInfo,
msg: InstantiateMsg,
) -> StdResult<Response> {
// 验证输入参数
if msg.name.is_empty() {
return Err(StdError::generic_err("Token name cannot be empty"));
}
if msg.symbol.is_empty() {
return Err(StdError::generic_err("Token symbol cannot be empty"));
}
if msg.decimals > 18 {
return Err(StdError::generic_err("Decimals cannot exceed 18"));
}
if msg.agent_did.is_empty() {
return Err(StdError::generic_err("Agent DID cannot be empty"));
}
// 调用标准 CW20 实例化
let cw20_msg = cw20_base::msg::InstantiateMsg {
name: msg.name.clone(),
symbol: msg.symbol.clone(),
decimals: msg.decimals,
initial_balances: msg.initial_balances.clone(),
mint: msg.mint.clone(),
marketing: None,
};
cw20_base::contract::instantiate(deps, env.clone(), info.clone(), cw20_msg)?;
// 验证管理员地址
let admin = deps.api.addr_validate(&msg.admin)?;
// 存储 Agent 元数据
let metadata = TokenMetadata {
agent_did: msg.agent_did.clone(),
created_at: env.block.height,
admin: admin.clone(),
token_type: msg.token_type.clone(),
paused: false,
cap: None,
};
TOKEN_METADATA.save(deps.storage, &metadata)?;
// DID 反向索引
AGENT_DID_LOOKUP.save(deps.storage, &msg.agent_did, &admin)?;
PAUSED.save(deps.storage, &false)?;
Ok(Response::new()
.add_attribute("action", "instantiate")
.add_attribute("agent_did", msg.agent_did)
.add_attribute("token_type", msg.token_type)
.add_attribute("admin", admin.to_string())
.add_attribute("name", msg.name)
.add_attribute("symbol", msg.symbol))
}
#[entry_point]
pub fn execute(
deps: DepsMut,
env: Env,
info: MessageInfo,
msg: ExecuteMsg,
) -> StdResult<Response> {
// 检查合约是否暂停
if PAUSED.load(deps.storage)? {
return Err(StdError::generic_err("Contract is paused"));
}
match msg {
ExecuteMsg::Transfer { recipient, amount } => {
execute_transfer(deps, env, info, recipient, amount)
}
ExecuteMsg::TransferFrom { owner, recipient, amount } => {
execute_transfer_from(deps, env, info, owner, recipient, amount)
}
ExecuteMsg::Approve { spender, amount } => {
execute_approve(deps, env, info, spender, amount)
}
ExecuteMsg::Mint { recipient, amount } => {
// 验证铸造权限
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Only admin can mint tokens"));
}
execute_mint(deps, env, info, recipient, amount)
}
ExecuteMsg::Burn { amount } => {
execute_burn(deps, env, info, amount)
}
ExecuteMsg::Send { contract, amount, msg } => {
execute_send(deps, env, info, contract, amount, msg)
}
ExecuteMsg::UpdateMetadata { new_did, new_admin } => {
execute_update_metadata(deps, env, info, new_did, new_admin)
}
ExecuteMsg::Pause {} => {
execute_pause(deps, env, info)
}
ExecuteMsg::Unpause {} => {
execute_unpause(deps, env, info)
}
ExecuteMsg::BatchTransfer { recipients, amounts } => {
execute_batch_transfer(deps, env, info, recipients, amounts)
}
}
}
#[entry_point]
pub fn query(deps: Deps, env: Env, msg: QueryMsg) -> StdResult<Binary> {
match msg {
QueryMsg::Balance { address } => query_balance(deps, env, address),
QueryMsg::TokenInfo {} => query_token_info(deps, env),
QueryMsg::Minter {} => query_minter(deps, env),
QueryMsg::Allowance { owner, spender } => {
query_allowance(deps, env, owner, spender)
}
QueryMsg::AllAllowances { owner, start_after, limit } => {
query_all_allowances(deps, env, owner, start_after, limit)
}
QueryMsg::AgentMetadata {} => query_agent_metadata(deps),
QueryMsg::TokenType {} => query_token_type(deps),
QueryMsg::IsPaused {} => query_is_paused(deps),
}
}
2.6 执行逻辑 (execute.rs)
use cosmwasm_std::{
Addr, DepsMut, Env, MessageInfo, Response, StdError, StdResult, Uint128,
};
use crate::state::{TokenMetadata, TOKEN_METADATA, AGENT_DID_LOOKUP, PAUSED};
/// 更新 Agent 元数据
pub fn execute_update_metadata(
deps: DepsMut,
_env: Env,
info: MessageInfo,
new_did: Option<String>,
new_admin: Option<String>,
) -> StdResult<Response> {
let mut metadata = TOKEN_METADATA.load(deps.storage)?;
// 仅管理员可更新
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized: admin only"));
}
let mut attrs = vec![("action", "update_metadata")];
if let Some(did) = new_did {
// 删除旧的 DID 索引
AGENT_DID_LOOKUP.remove(deps.storage, &metadata.agent_did);
// 更新 DID
AGENT_DID_LOOKUP.save(deps.storage, &did, &info.sender)?;
metadata.agent_did = did.clone();
attrs.push(("new_did", &did));
}
if let Some(admin) = new_admin {
let new_admin_addr = deps.api.addr_validate(&admin)?;
metadata.admin = new_admin_addr;
attrs.push(("new_admin", &admin));
}
TOKEN_METADATA.save(deps.storage, &metadata)?;
Ok(Response::new().add_attributes(attrs))
}
/// 暂停合约（仅管理员）
pub fn execute_pause(
deps: DepsMut,
_env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized: admin only"));
}
PAUSED.save(deps.storage, &true)?;
Ok(Response::new().add_attribute("action", "pause"))
}
/// 恢复合约（仅管理员）
pub fn execute_unpause(
deps: DepsMut,
_env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized: admin only"));
}
PAUSED.save(deps.storage, &false)?;
Ok(Response::new().add_attribute("action", "unpause"))
}
/// 批量转账
pub fn execute_batch_transfer(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipients: Vec<String>,
amounts: Vec<Uint128>,
) -> StdResult<Response> {
if recipients.len() != amounts.len() {
return Err(StdError::generic_err(
"Recipients and amounts length mismatch",
));
}
if recipients.is_empty() {
return Err(StdError::generic_err("Empty batch transfer"));
}
let mut total = Uint128::zero();
for amount in &amounts {
total = total.checked_add(*amount)?;
}
// 先燃烧发送者的代币
cw20_base::contract::execute_burn(deps.branch(), env.clone(), info.clone(), total)?;
// 为每个接收者铸造
let mut res = Response::new()
.add_attribute("action", "batch_transfer")
.add_attribute("sender", info.sender)
.add_attribute("count", recipients.len().to_string());
for (recipient, amount) in recipients.iter().zip(amounts.iter()) {
cw20_base::contract::execute_mint(
deps.branch(),
env.clone(),
info.clone(),
recipient.clone(),
*amount,
)?;
res = res.add_attribute(
format!("transfer_{}", recipient),
amount.to_string(),
);
}
Ok(res)
}
2.7 查询逻辑 (query.rs)
use cosmwasm_std::{to_binary, Binary, Deps, StdResult};
use crate::state::TOKEN_METADATA;
/// 查询 Agent 元数据
pub fn query_agent_metadata(deps: Deps) -> StdResult<Binary> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
to_binary(&metadata)
}
/// 查询代币类型
pub fn query_token_type(deps: Deps) -> StdResult<Binary> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
to_binary(&metadata.token_type)
}
/// 查询暂停状态
pub fn query_is_paused(deps: Deps) -> StdResult<Binary> {
let paused = PAUSED.load(deps.storage)?;
to_binary(&paused)
}
2.8 合约部署流程
#!/usr/bin/env python3
"""
Agent 代币合约部署脚本
用法: python3 deploy_token.py --agent-did "did:msg:agent:abc123"
"""
import json
import subprocess
import argparse
import hashlib
from pathlib import Path
def compile_contract(contract_path: str) -> str:
"""使用 cosmwasm-opt 编译合约"""
cmd = [
"docker", "run", "--rm", "-v",
f"{contract_path}:/code",
"--platform", "linux/amd64",
"cosmwasm/optimizer:0.16.0",
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Compilation failed: {result.stderr}")
wasm_path = Path(contract_path) / "artifacts" / "agent_token.wasm"
return str(wasm_path)
def store_contract(wasm_path: str, wallet: str) -> str:
"""在 MSG Chain 上存储合约代码"""
cmd = [
"msgcli", "tx", "wasm", "store", wasm_path,
"--from", wallet,
"--gas", "auto",
"--gas-adjustment", "1.3",
"--yes",
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Store failed: {result.stderr}")
# 解析返回的 Code ID
for line in result.stdout.split("\n"):
if "code_id" in line:
return line.split(":")[-1].strip()
raise RuntimeError("Could not parse code_id")
def instantiate_contract(
code_id: str,
wallet: str,
agent_did: str,
name: str,
symbol: str,
decimals: int,
initial_supply: int,
admin: str,
token_type: str = "utility",
) -> str:
"""实例化合约"""
init_msg = {
"name": name,
"symbol": symbol,
"decimals": decimals,
"initial_balances": [
{
"address": wallet,
"amount": str(initial_supply),
}
],
"mint": {
"minter": wallet,
"cap": None,
},
"agent_did": agent_did,
"token_type": token_type,
"admin": admin,
}
cmd = [
"msgcli", "tx", "wasm", "instantiate",
code_id,
json.dumps(init_msg),
"--from", wallet,
"--label", f"{symbol}-{agent_did[:8]}",
"--admin", admin,
"--gas", "auto",
"--gas-adjustment", "1.3",
"--yes",
]
result = subprocess.run(cmd, capture_output=True, text=True)
if result.returncode != 0:
raise RuntimeError(f"Instantiate failed: {result.stderr}")
for line in result.stdout.split("\n"):
if "contract_address" in line:
return line.split(":")[-1].strip()
raise RuntimeError("Could not parse contract address")
def main():
parser = argparse.ArgumentParser(
description="Deploy Agent CW20 Token Contract"
)
parser.add_argument("--agent-did", required=True, help="Agent DID")
parser.add_argument("--name", required=True, help="Token name")
parser.add_argument("--symbol", required=True, help="Token symbol")
parser.add_argument("--decimals", type=int, default=6)
parser.add_argument("--supply", type=int, default=1_000_000)
parser.add_argument("--type", default="utility",
choices=["utility", "governance", "reward", "social"])
parser.add_argument("--admin", required=True, help="Admin address")
parser.add_argument("--wallet", default="agent_admin",
help="Local wallet name")
parser.add_argument("--contract-path", default="./agent-token",
help="Contract source path")
args = parser.parse_args()
print(f"Compiling contract from {args.contract_path}...")
wasm = compile_contract(args.contract_path)
print(f"Compiled: {wasm}")
print("Storing contract on MSG Chain...")
code_id = store_contract(wasm, args.wallet)
print(f"Code ID: {code_id}")
print("Instantiating contract...")
contract_addr = instantiate_contract(
code_id=code_id,
wallet=args.wallet,
agent_did=args.agent_did,
name=args.name,
symbol=args.symbol,
decimals=args.decimals,
initial_supply=args.supply,
admin=args.admin,
token_type=args.type,
)
print(f"Contract address: {contract_addr}")
print(f"Token {args.symbol} ({args.name}) deployed successfully!")
return contract_addr
if __name__ == "__main__":
main()
2.9 合约迁移逻辑
use cosmwasm_std::{entry_point, DepsMut, Env, StdResult, Response};
use crate::msg::MigrateMsg;
#[entry_point]
pub fn migrate(deps: DepsMut, _env: Env, msg: MigrateMsg) -> StdResult<Response> {
// 更新合约版本
deps.storage.set(
b"contract_version",
msg.new_contract_version.as_bytes(),
);
Ok(Response::new()
.add_attribute("action", "migrate")
.add_attribute("version", msg.new_contract_version))
}
2.10 合约测试
#[cfg(test)]
mod tests {
use cosmwasm_std::testing::{
mock_dependencies, mock_env, mock_info, MockApi, MockQuerier,
};
use cosmwasm_std::{coins, from_binary, Addr, Uint128};
use cw20::{BalanceResponse, Cw20Coin, MinterResponse, TokenInfoResponse};
use crate::contract::{instantiate, execute, query};
use crate::msg::{ExecuteMsg, InstantiateMsg, QueryMsg};
const ADMIN: &str = "msg1agentadmin000000000000000000000000000";
const USER1: &str = "msg1user1000000000000000000000000000000000";
const AGENT_DID: &str = "did:msg:agent:test001";
fn setup_contract(deps: DepsMut) {
let msg = InstantiateMsg {
name: "Agent Token".to_string(),
symbol: "AGT".to_string(),
decimals: 6,
initial_balances: vec![Cw20Coin {
address: ADMIN.to_string(),
amount: Uint128::new(1_000_000),
}],
mint: Some(MinterResponse {
minter: ADMIN.to_string(),
cap: None,
}),
agent_did: AGENT_DID.to_string(),
token_type: "utility".to_string(),
admin: ADMIN.to_string(),
};
let env = mock_env();
let info = mock_info(ADMIN, &[]);
let res = instantiate(deps, env, info, msg).unwrap();
assert_eq!(res.attributes[0].value, "instantiate");
}
#[test]
fn proper_initialization() {
let mut deps = mock_dependencies();
setup_contract(deps.as_mut());
let env = mock_env();
let res = query(
deps.as_ref(),
env.clone(),
QueryMsg::Balance {
address: ADMIN.to_string(),
},
)
.unwrap();
let balance: BalanceResponse = from_binary(&res).unwrap();
assert_eq!(balance.balance, Uint128::new(1_000_000));
let res = query(
deps.as_ref(),
env,
QueryMsg::TokenInfo {},
)
.unwrap();
let info: TokenInfoResponse = from_binary(&res).unwrap();
assert_eq!(info.name, "Agent Token");
assert_eq!(info.symbol, "AGT");
assert_eq!(info.decimals, 6);
}
#[test]
fn transfer_works() {
let mut deps = mock_dependencies();
setup_contract(deps.as_mut());
let env = mock_env();
let info = mock_info(ADMIN, &[]);
let msg = ExecuteMsg::Transfer {
recipient: USER1.to_string(),
amount: Uint128::new(100),
};
execute(deps.as_mut(), env.clone(), info, msg).unwrap();
let res = query(
deps.as_ref(),
env,
QueryMsg::Balance {
address: USER1.to_string(),
},
)
.unwrap();
let balance: BalanceResponse = from_binary(&res).unwrap();
assert_eq!(balance.balance, Uint128::new(100));
}
#[test]
fn mint_only_by_admin() {
let mut deps = mock_dependencies();
setup_contract(deps.as_mut());
let env = mock_env();
let info = mock_info(USER1, &[]);
let msg = ExecuteMsg::Mint {
recipient: USER1.to_string(),
amount: Uint128::new(1000),
};
let err = execute(deps.as_mut(), env, info, msg).unwrap_err();
assert_eq!(err.to_string(), "Only admin can mint tokens");
}
#[test]
fn pause_unpause_works() {
let mut deps = mock_dependencies();
setup_contract(deps.as_mut());
// 暂停
let env = mock_env();
let info = mock_info(ADMIN, &[]);
execute(
deps.as_mut(),
env.clone(),
info,
ExecuteMsg::Pause {},
)
.unwrap();
// 暂停后转账应失败
let info = mock_info(ADMIN, &[]);
let msg = ExecuteMsg::Transfer {
recipient: USER1.to_string(),
amount: Uint128::new(50),
};
let err = execute(deps.as_mut(), env.clone(), info, msg).unwrap_err();
assert_eq!(err.to_string(), "Contract is paused");
// 恢复
let info = mock_info(ADMIN, &[]);
execute(
deps.as_mut(),
env,
info,
ExecuteMsg::Unpause {},
)
.unwrap();
}
#[test]
fn batch_transfer_works() {
let mut deps = mock_dependencies();
setup_contract(deps.as_mut());
let env = mock_env();
let info = mock_info(ADMIN, &[]);
let msg = ExecuteMsg::BatchTransfer {
recipients: vec![
USER1.to_string(),
"msg1user2000000000000000000000000000000000".to_string(),
],
amounts: vec![
Uint128::new(200),
Uint128::new(300),
],
};
execute(deps.as_mut(), env.clone(), info, msg).unwrap();
let res = query(
deps.as_ref(),
env.clone(),
QueryMsg::Balance {
address: USER1.to_string(),
},
)
.unwrap();
let balance: BalanceResponse = from_binary(&res).unwrap();
assert_eq!(balance.balance, Uint128::new(200));
}
#[test]
fn invalid_instantiation_fails() {
let mut deps = mock_dependencies();
let msg = InstantiateMsg {
name: "".to_string(),
symbol: "AGT".to_string(),
decimals: 6,
initial_balances: vec![],
mint: None,
agent_did: "did:msg:agent:test".to_string(),
token_type: "utility".to_string(),
admin: ADMIN.to_string(),
};
let env = mock_env();
let info = mock_info(ADMIN, &[]);
let err = instantiate(deps.as_mut(), env, info, msg).unwrap_err();
assert_eq!(err.to_string(), "Token name cannot be empty");
}
}
3. Token 分发策略
3.1 分发计划设计
合理的代币分发是代币经济成功的关键。AI Agent 的标准分发方案如下：
代币总供应量: 10,000,000 AGT
├── 社区销售: 30% (3,000,000) — 公开发售
├── 生态基金: 25% (2,500,000) — 生态建设与激励
├── 团队归属: 20% (2,000,000) — 4年线性解锁
├── 流动性: 15% (1,500,000) — DEX 流动性池
└── 空投: 10% (1,000,000) — 早期用户与贡献者
3.2 分发管理器
#!/usr/bin/env python3
"""
Token 分发管理器
支持空投、社区销售、归属计划
"""
import json
import time
import asyncio
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from decimal import Decimal
@dataclass
class DistributionConfig:
"""分发配置"""
total_supply: int
community_sale_pct: float = 0.30
ecosystem_fund_pct: float = 0.25
team_vesting_pct: float = 0.20
liquidity_pct: float = 0.15
airdrop_pct: float = 0.10
@property
def community_sale_amount(self) -> int:
return int(self.total_supply * self.community_sale_pct)
@property
def ecosystem_fund_amount(self) -> int:
return int(self.total_supply * self.ecosystem_fund_pct)
@property
def team_vesting_amount(self) -> int:
return int(self.total_supply * self.team_vesting_pct)
@property
def liquidity_amount(self) -> int:
return int(self.total_supply * self.liquidity_pct)
@property
def airdrop_amount(self) -> int:
return int(self.total_supply * self.airdrop_pct)
class TokenDistributor:
"""Token 分发器"""
def __init__(
self,
token_addr: str,
admin_key: str,
client: "MsgChainClient",
):
self.token_addr = token_addr
self.admin_key = admin_key
self.client = client
self.gas_limit = 2000000
async def airdrop(
self,
recipients: List[str],
amounts: List[int],
memo: str = "Agent token airdrop",
batch_size: int = 50,
) -> List[str]:
"""批量空投
Args:
recipients: 接收者地址列表
amounts: 对应数量列表
memo: 交易备注
batch_size: 每批地址数
Returns:
交易哈希列表
"""
if len(recipients) != len(amounts):
raise ValueError("Recipients and amounts length mismatch")
if not recipients:
return []
tx_hashes = []
# 分批处理
for i in range(0, len(recipients), batch_size):
batch_r = recipients[i:i + batch_size]
batch_a = amounts[i:i + batch_size]
msgs = []
for addr, amount in zip(batch_r, batch_a):
msg = {
"contract_addr": self.token_addr,
"msg": {
"transfer": {
"recipient": addr,
"amount": str(amount),
}
},
"funds": [],
}
msgs.append(msg)
tx_hash = await self.client.execute_contract_msgs(
msgs, self.admin_key, memo, self.gas_limit
)
tx_hashes.append(tx_hash)
print(f"Batch {i // batch_size + 1}: {tx_hash}")
# 限速，避免链过载
if i + batch_size < len(recipients):
await asyncio.sleep(1)
return tx_hashes
async def airdrop_from_csv(
self,
csv_path: str,
amount_column: str = "amount",
addr_column: str = "address",
max_amount: Optional[int] = None,
) -> List[str]:
"""从 CSV 文件读取空投列表
CSV 格式示例:
address,amount
msg1abc...,1000
msg1def...,2000
"""
import csv
recipients = []
amounts = []
with open(csv_path, "r") as f:
reader = csv.DictReader(f)
for row in reader:
addr = row[addr_column].strip()
amount = int(row[amount_column].strip())
if not addr.startswith("msg1"):
print(f"Skipping non-msg address: {addr}")
continue
if max_amount and amount > max_amount:
amount = max_amount
recipients.append(addr)
amounts.append(amount)
print(f"Loaded {len(recipients)} recipients from {csv_path}")
return await self.airdrop(recipients, amounts)
async def community_sale(
self,
participants: List[Tuple[str, int, int]], # (addr, amount, msg_amount)
price_per_token: int, # 每个代币对应的 umsg 数量
) -> str:
"""社区销售
Args:
participants: (地址, 代币数量, 支付的 umsg 数量)
price_per_token: 每个代币价格（umsg）
Returns:
交易哈希
"""
msgs = []
total_msg = 0
total_tokens = 0
for addr, token_amount, msg_amount in participants:
expected_msg = token_amount * price_per_token
if msg_amount != expected_msg:
raise ValueError(
f"Payment mismatch for {addr}: "
f"expected {expected_msg}, got {msg_amount}"
)
msgs.append({
"contract_addr": self.token_addr,
"msg": {
"transfer": {
"recipient": addr,
"amount": str(token_amount),
}
},
"funds": [{"denom": "umsg", "amount": str(msg_amount)}],
})
total_msg += msg_amount
total_tokens += token_amount
print(f"Sale: {total_tokens} tokens for {total_msg} umsg")
return await self.client.execute_contract_msgs(
msgs, self.admin_key,
"Agent token community sale",
self.gas_limit * len(msgs),
)
async def create_vesting_schedule(
self,
beneficiary: str,
total_amount: int,
duration_days: int = 1460, # 默认 4 年
cliff_days: int = 365, # 默认 1 年悬崖期
start_timestamp: Optional[int] = None,
) -> dict:
"""创建线性解锁计划
使用简单的线性解锁公式:
unlockable = total * min(1, elapsed / duration)
Args:
beneficiary: 受益地址
total_amount: 总解锁数量
duration_days: 解锁周期（天）
cliff_days: 悬崖期（天）
start_timestamp: 起始时间戳
Returns:
Vesting 计划信息
"""
if start_timestamp is None:
start_timestamp = int(time.time())
schedule = {
"beneficiary": beneficiary,
"total_amount": total_amount,
"start_timestamp": start_timestamp,
"duration_seconds": duration_days * 86400,
"cliff_seconds": cliff_days * 86400,
"released_amount": 0,
"last_claim_timestamp": start_timestamp,
}
# 将归属计划存储到合约或本地
await self._store_vesting_schedule(beneficiary, schedule)
print(f"Vesting created for {beneficiary}:")
print(f" Total: {total_amount}")
print(f" Duration: {duration_days} days")
print(f" Cliff: {cliff_days} days")
return schedule
async def _store_vesting_schedule(
self, beneficiary: str, schedule: dict
) -> None:
"""存储归属计划到链上合约"""
msg = {
"contract_addr": self.token_addr,
"msg": {
"create_vesting": {
"beneficiary": beneficiary,
"total_amount": str(schedule["total_amount"]),
"start_timestamp": str(schedule["start_timestamp"]),
"duration_seconds": str(schedule["duration_seconds"]),
"cliff_seconds": str(schedule["cliff_seconds"]),
}
},
"funds": [],
}
await self.client.execute_contract_msg(
msg, self.admin_key,
f"Create vesting for {beneficiary}",
)
async def claim_vested(self, beneficiary: str) -> int:
"""领取已解锁的代币
Returns:
实际领取数量
"""
schedule = await self._get_vesting_schedule(beneficiary)
now = int(time.time())
if now < schedule["start_timestamp"] + schedule["cliff_seconds"]:
print(f"Still in cliff period for {beneficiary}")
return 0
elapsed = now - schedule["start_timestamp"]
duration = schedule["duration_seconds"]
# 线性解锁
unlocked_pct = min(Decimal(elapsed) / Decimal(duration), Decimal("1"))
total_unlocked = int(
Decimal(schedule["total_amount"]) * unlocked_pct
)
claimable = total_unlocked - schedule["released_amount"]
if claimable <= 0:
print(f"No claimable tokens for {beneficiary}")
return 0
# 执行领取
msg = {
"contract_addr": self.token_addr,
"msg": {
"transfer": {
"recipient": beneficiary,
"amount": str(claimable),
}
},
"funds": [],
}
await self.client.execute_contract_msg(
msg, self.admin_key,
f"Vesting claim for {beneficiary}",
)
# 更新本地记录
schedule["released_amount"] += claimable
schedule["last_claim_timestamp"] = now
await self._store_vesting_schedule(beneficiary, schedule)
print(f"Claimed {claimable} tokens for {beneficiary}")
return claimable
async def _get_vesting_schedule(self, beneficiary: str) -> dict:
"""从链上查询归属计划"""
query = {
"vesting_schedule": {
"beneficiary": beneficiary,
}
}
return await self.client.query_contract(self.token_addr, query)
async def distribute_all(
self,
config: DistributionConfig,
airdrop_list: List[Tuple[str, int]],
sale_participants: List[Tuple[str, int, int]],
team_members: List[Tuple[str, int]],
dex_addr: str,
) -> dict:
"""完整分发流程"""
result = {}
# 1. 空投
print("\n=== 开始空投 ===")
airdrop_addrs = [a for a, _ in airdrop_list]
airdrop_amts = [a for _, a in airdrop_list]
result["airdrop_tx"] = await self.airdrop(
airdrop_addrs, airdrop_amts
)
# 2. 社区销售
print("\n=== 社区销售 ===")
result["sale_tx"] = await self.community_sale(
sale_participants, price_per_token=100 # 100 umsg = 1 token
)
# 3. 团队归属
print("\n=== 团队归属 ===")
result["vesting_schedules"] = []
for addr, amount in team_members:
schedule = await self.create_vesting_schedule(
addr, amount,
duration_days=1460,
cliff_days=365,
)
result["vesting_schedules"].append(schedule)
# 4. 流动性
print("\n=== 注入流动性 ===")
result["liquidity_tx"] = await self._add_liquidity_to_dex(
dex_addr, config.liquidity_amount
)
# 5. 生态基金
print("\n=== 生态基金 ===")
result["ecosystem_fund_tx"] = await self._fund_ecosystem(
config.ecosystem_fund_amount
)
return result
async def _add_liquidity_to_dex(
self, dex_addr: str, amount: int
) -> str:
"""向 DEX 添加流动性"""
msg = {
"contract_addr": dex_addr,
"msg": {
"add_liquidity": {
"token_amount": str(amount),
"msg_amount": str(amount // 2),
}
},
"funds": [{"denom": "umsg", "amount": str(amount // 2)}],
}
return await self.client.execute_contract_msg(
msg, self.admin_key,
f"Add liquidity: {amount} tokens",
)
async def _fund_ecosystem(self, amount: int) -> str:
"""向生态基金转入代币"""
msg = {
"contract_addr": self.token_addr,
"msg": {
"transfer": {
"recipient": "msg1ecosystemfund00000000000000000000000",
"amount": str(amount),
}
},
"funds": [],
}
return await self.client.execute_contract_msg(
msg, self.admin_key,
"Fund ecosystem reserve",
)
3.3 Vesting 合约实现
use cosmwasm_std::{
Addr, DepsMut, Env, MessageInfo, Response, StdError,
StdResult, Storage, Uint128,
};
use cw_storage_plus::Map;
use serde::{Deserialize, Serialize};
/// 归属计划
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct VestingSchedule {
pub beneficiary: Addr,
pub total_amount: Uint128,
pub start_time: u64,
pub duration: u64,
pub cliff: u64,
pub released: Uint128,
}
pub const VESTING_SCHEDULES: Map<&Addr, VestingSchedule> =
Map::new("vesting");
/// 创建归属计划（仅管理员）
pub fn create_vesting(
deps: DepsMut,
_env: Env,
info: MessageInfo,
beneficiary: String,
total_amount: Uint128,
start_time: u64,
duration: u64,
cliff: u64,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized"));
}
let beneficiary_addr = deps.api.addr_validate(&beneficiary)?;
if VESTING_SCHEDULES.has(deps.storage, &beneficiary_addr) {
return Err(StdError::generic_err(
"Vesting schedule already exists for this beneficiary",
));
}
if cliff > duration {
return Err(StdError::generic_err(
"Cliff cannot exceed duration",
));
}
let schedule = VestingSchedule {
beneficiary: beneficiary_addr.clone(),
total_amount,
start_time,
duration,
cliff,
released: Uint128::zero(),
};
VESTING_SCHEDULES.save(deps.storage, &beneficiary_addr, &schedule)?;
Ok(Response::new()
.add_attribute("action", "create_vesting")
.add_attribute("beneficiary", beneficiary)
.add_attribute("total_amount", total_amount)
.add_attribute("duration", duration.to_string())
.add_attribute("cliff", cliff.to_string()))
}
/// 计算可领取数量
pub fn compute_claimable(
schedule: &VestingSchedule,
current_time: u64,
) -> Uint128 {
if current_time < schedule.start_time + schedule.cliff {
return Uint128::zero();
}
let elapsed = current_time
.saturating_sub(schedule.start_time)
.min(schedule.duration);
// 线性解锁: (total * elapsed) / duration
let unlocked = schedule
.total_amount
.multiply_ratio(elapsed, schedule.duration);
unlocked.saturating_sub(schedule.released)
}
/// 领取已解锁代币
pub fn claim_vested(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let schedule = VESTING_SCHEDULES
.load(deps.storage, &info.sender)?;
let claimable = compute_claimable(&schedule, env.block.time.seconds());
if claimable.is_zero() {
return Err(StdError::generic_err("No claimable tokens"));
}
// 铸造待领取的代币
cw20_base::contract::execute_mint(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
claimable,
)?;
// 更新已释放数量
let mut updated = schedule.clone();
updated.released = schedule.released.checked_add(claimable)?;
VESTING_SCHEDULES.save(deps.storage, &info.sender, &updated)?;
Ok(Response::new()
.add_attribute("action", "claim_vested")
.add_attribute("beneficiary", info.sender)
.add_attribute("amount", claimable))
}
3.4 批量归属创建脚本
#!/usr/bin/env python3
"""
批量创建团队归属计划
"""
import json
import sys
from typing import List, Tuple
import asyncio
async def create_team_vesting(
distributor: TokenDistributor,
members_file: str,
total_pool: int,
duration_days: int = 1460,
cliff_days: int = 365,
):
"""从 JSON 文件读取团队成员列表并创建归属
members.json 格式:
{
"members": [
{"address": "msg1...", "name": "Alice", "weight": 30},
{"address": "msg1...", "name": "Bob", "weight": 25},
{"address": "msg1...", "name": "Charlie", "weight": 20}
]
}
"""
with open(members_file, "r") as f:
data = json.load(f)
members = data["members"]
total_weight = sum(m["weight"] for m in members)
print(f"Creating vesting for {len(members)} team members")
print(f"Total pool: {total_pool}")
print(f"Total weight: {total_weight}")
schedules = []
for member in members:
allocation = int(total_pool * member["weight"] / total_weight)
schedule = await distributor.create_vesting_schedule(
beneficiary=member["address"],
total_amount=allocation,
duration_days=duration_days,
cliff_days=cliff_days,
)
schedules.append({
"name": member["name"],
"address": member["address"],
"amount": allocation,
"schedule": schedule,
})
print(f" {member['name']}: {allocation} tokens")
# 输出汇总
print("\n=== Vesting Summary ===")
for s in schedules:
print(f"{s['name']:20} {s['address']:50} {s['amount']:10}")
print(f"\nTotal allocated: {sum(s['amount'] for s in schedules)}")
return schedules
if __name__ == "__main__":
# 从环境变量读取配置
TOKEN_ADDR = sys.argv[1]
MEMBERS_FILE = sys.argv[2]
TOTAL_POOL = int(sys.argv[3]) if len(sys.argv) > 3 else 2_000_000
distributor = TokenDistributor(
token_addr=TOKEN_ADDR,
admin_key="agent_admin",
client=None, # 注入实际 client
)
asyncio.run(create_team_vesting(
distributor, MEMBERS_FILE, TOTAL_POOL
))
4. 质押与奖励
4.1 质押机制设计
质押模块允许用户锁定代币以换取奖励，是提升代币需求的重要机制。
用户流程:
1. 用户调用 stake(amount)
2. 合约锁定用户的代币
3. 每区块按 reward_rate 累积奖励
4. 用户随时调用 claim_rewards() 领取
5. 用户调用 unstake(amount) 进入解绑期
6. 解绑期结束后用户领取代币
4.2 质押合约核心实现
use cosmwasm_std::{
Addr, Deps, DepsMut, Env, MessageInfo, Response,
StdError, StdResult, Storage, Uint128,
};
use cw_storage_plus::{Item, Map};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
// ===== 状态 =====
/// 质押者信息
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct StakerInfo {
pub staked: Uint128,
pub reward_debt: Uint128,
pub staked_at: u64,
}
/// 全局质押状态
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct StakingState {
pub total_staked: Uint128,
pub reward_rate: Uint128, // 每区块奖励数量
pub reward_index: Uint128, // 累积奖励指数
pub last_update_block: u64, // 上次更新时间
pub min_stake: Uint128, // 最低质押量
pub unstaking_period: u64, // 解绑周期（秒）
}
pub const STAKING_STATE: Item<StakingState> = Item::new("staking_state");
pub const STAKERS: Map<&Addr, StakerInfo> = Map::new("stakers");
pub const UNSTAKING_REQUESTS: Map<(&Addr, u64), UnstakingRequest> =
Map::new("unstaking");
/// 解绑请求
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct UnstakingRequest {
pub staker: Addr,
pub amount: Uint128,
pub request_time: u64,
pub complete_time: u64,
}
// ===== 初始化 =====
pub fn init_staking(
deps: DepsMut,
reward_rate: Uint128,
min_stake: Uint128,
unstaking_period: u64,
) -> StdResult<Response> {
let state = StakingState {
total_staked: Uint128::zero(),
reward_rate,
reward_index: Uint128::zero(),
last_update_block: 0,
min_stake,
unstaking_period,
};
STAKING_STATE.save(deps.storage, &state)?;
Ok(Response::new().add_attribute("action", "init_staking"))
}
// ===== 核心函数 =====
/// 更新全局奖励指数
pub fn update_reward_index(
storage: &mut dyn Storage,
env: &Env,
) -> StdResult<()> {
let mut state = STAKING_STATE.load(storage)?;
if state.total_staked.is_zero() {
state.last_update_block = env.block.height;
STAKING_STATE.save(storage, &state)?;
return Ok(());
}
let blocks_elapsed = env.block.height - state.last_update_block;
if blocks_elapsed == 0 {
return Ok(());
}
// reward_index += (reward_rate * blocks_elapsed) / total_staked
let reward_per_staked = state
.reward_rate
.checked_mul(Uint128::from(blocks_elapsed))?
.checked_div(state.total_staked)?;
state.reward_index = state.reward_index.checked_add(reward_per_staked)?;
state.last_update_block = env.block.height;
STAKING_STATE.save(storage, &state)?;
Ok(())
}
/// 更新单个质押者的奖励债务
pub fn update_staker_reward(
storage: &mut dyn Storage,
staker_addr: &Addr,
) -> StdResult<()> {
let state = STAKING_STATE.load(storage)?;
let mut staker = STAKERS.load(storage, staker_addr)?;
// owed = staker.staked * (reward_index - staker.reward_debt)
let reward_diff = state
.reward_index
.checked_sub(staker.reward_debt)?;
let owed = staker.staked.checked_mul(reward_diff)?;
staker.reward_debt = state.reward_index;
STAKERS.save(storage, staker_addr, &staker)?;
// 累积奖励
let mut rewards = REWARD_POOL
.load(storage)?
.reward_balances
.get(staker_addr)
.copied()
.unwrap_or_default();
rewards = rewards.checked_add(owed)?;
let mut pool = REWARD_POOL.load(storage)?;
pool.reward_balances.insert(staker_addr.clone(), rewards);
REWARD_POOL.save(storage, &pool)?;
Ok(())
}
// ===== 执行函数 =====
#[entry_point]
pub fn stake(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let state = STAKING_STATE.load(deps.storage)?;
if amount < state.min_stake {
return Err(StdError::generic_err(format!(
"Minimum stake is {}",
state.min_stake
)));
}
// 更新奖励指数
update_reward_index(deps.storage, &env)?;
// 更新用户奖励（如果有历史质押）
if STAKERS.has(deps.storage, &info.sender) {
update_staker_reward(deps.storage, &info.sender)?;
}
// 从用户地址转移代币到合约
cw20_base::contract::execute_transfer_from(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
env.contract.address.to_string(),
amount,
)?;
// 更新质押者信息
let mut staker = STAKERS
.load(deps.storage, &info.sender)
.unwrap_or(StakerInfo {
staked: Uint128::zero(),
reward_debt: state.reward_index,
staked_at: env.block.time.seconds(),
});
staker.staked = staker.staked.checked_add(amount)?;
// 新质押者的 reward_debt 设为当前指数
staker.reward_debt = state.reward_index;
STAKERS.save(deps.storage, &info.sender, &staker)?;
// 更新总量
let mut current_state = STAKING_STATE.load(deps.storage)?;
current_state.total_staked =
current_state.total_staked.checked_add(amount)?;
STAKING_STATE.save(deps.storage, ¤t_state)?;
Ok(Response::new()
.add_attribute("action", "stake")
.add_attribute("staker", info.sender)
.add_attribute("amount", amount)
.add_attribute("total_staked", current_state.total_staked))
}
#[entry_point]
pub fn claim_rewards(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> StdResult<Response> {
update_reward_index(deps.storage, &env)?;
update_staker_reward(deps.storage, &info.sender)?;
let mut pool = REWARD_POOL.load(deps.storage)?;
let reward = pool
.reward_balances
.remove(&info.sender)
.unwrap_or_default();
if reward.is_zero() {
return Err(StdError::generic_err("No rewards to claim"));
}
// 发放奖励
cw20_base::contract::execute_mint(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
reward,
)?;
REWARD_POOL.save(deps.storage, &pool)?;
Ok(Response::new()
.add_attribute("action", "claim_rewards")
.add_attribute("staker", info.sender)
.add_attribute("rewards", reward))
}
#[entry_point]
pub fn unstake(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let state = STAKING_STATE.load(deps.storage)?;
update_reward_index(deps.storage, &env)?;
update_staker_reward(deps.storage, &info.sender)?;
let mut staker = STAKERS.load(deps.storage, &info.sender)?;
if staker.staked < amount {
return Err(StdError::generic_err("Insufficient staked amount"));
}
staker.staked = staker.staked.checked_sub(amount)?;
STAKERS.save(deps.storage, &info.sender, &staker)?;
// 更新总量
let mut current_state = STAKING_STATE.load(deps.storage)?;
current_state.total_staked =
current_state.total_staked.checked_sub(amount)?;
STAKING_STATE.save(deps.storage, ¤t_state)?;
// 创建解锁请求
let request = UnstakingRequest {
staker: info.sender.clone(),
amount,
request_time: env.block.time.seconds(),
complete_time: env.block.time.seconds() + state.unstaking_period,
};
UNSTAKING_REQUESTS.save(
deps.storage,
(&info.sender, env.block.time.seconds()),
&request,
)?;
Ok(Response::new()
.add_attribute("action", "unstake")
.add_attribute("staker", info.sender)
.add_attribute("amount", amount)
.add_attribute("unlock_time", request.complete_time.to_string()))
}
#[entry_point]
pub fn withdraw_unstaked(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let now = env.block.time.seconds();
// 查找所有已完成的解锁请求
let pending_requests: Vec<_> = UNSTAKING_REQUESTS
.prefix(&info.sender)
.range(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.filter(|(_, r)| r.complete_time <= now)
.collect::<StdResult<Vec<_>>>()?;
let total_withdraw: Uint128 = pending_requests
.iter()
.map(|(_, r)| r.amount)
.sum();
if total_withdraw.is_zero() {
return Err(StdError::generic_err(
"No completed unstaking requests",
));
}
// 删除已处理的请求
for (key, _) in &pending_requests {
UNSTAKING_REQUESTS.remove(deps.storage, (&info.sender, *key));
}
// 发送代币给用户
cw20_base::contract::execute_transfer(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
total_withdraw,
)?;
Ok(Response::new()
.add_attribute("action", "withdraw_unstaked")
.add_attribute("staker", info.sender)
.add_attribute("amount", total_withdraw))
}
4.3 奖励计算器 (Python)
#!/usr/bin/env python3
"""
质押奖励计算器
用于估算 APY 和收益
"""
from dataclasses import dataclass
from typing import Optional
@dataclass
class StakingConfig:
"""质押配置"""
total_staked: int
reward_rate_per_block: int # 每区块奖励代币数
blocks_per_year: int = 5256000 # MSG Chain 约 6 秒/块
your_stake: int = 0
@property
def rewards_per_year(self) -> int:
return self.reward_rate_per_block * self.blocks_per_year
@property
def apy(self) -> float:
"""估算年化收益率"""
if self.total_staked == 0:
return 0.0
return self.rewards_per_year / self.total_staked
def estimated_daily_reward(self, stake: Optional[int] = None) -> float:
"""估算每日收益"""
s = stake or self.your_stake
if self.total_staked == 0:
return 0.0
blocks_per_day = self.blocks_per_year / 365
daily_rewards = self.reward_rate_per_block * blocks_per_day
return s * daily_rewards / self.total_staked
def print_report(self):
"""打印详细报告"""
print("=" * 50)
print("Staking Reward Report")
print("=" * 50)
print(f"Total Staked: {self.total_staked:>15}")
print(f"Reward Rate (block): {self.reward_rate_per_block:>15}")
print(f"Rewards / Year: {self.rewards_per_year:>15}")
print(f"APY: {self.apy:>14.2%}")
print("-" * 50)
print(f"Your Stake: {self.your_stake:>15}")
print(f"Est. Daily Reward: {self.estimated_daily_reward():>14.2f}")
print(f"Est. Weekly Reward: {self.estimated_daily_reward() * 7:>14.2f}")
print(f"Est. Monthly Reward: {self.estimated_daily_reward() * 30:>14.2f}")
print(f"Est. Yearly Reward: {self.estimated_daily_reward() * 365:>14.2f}")
print("=" * 50)
def compound_apy(
daily_reward: float,
current_stake: int,
days: int = 365,
) -> float:
"""计算复利 APY（假设每日复投）"""
daily_rate = daily_reward / current_stake if current_stake > 0 else 0
return (1 + daily_rate) ** days - 1
def find_required_stake(
target_daily_reward: float,
total_staked: int,
reward_rate: int,
blocks_per_year: int = 5256000,
) -> int:
"""计算达到目标每日收益所需的质押量"""
total_rewards_per_day = (
reward_rate * blocks_per_year / 365
)
if total_rewards_per_day == 0:
return 0
ratio = target_daily_reward / total_rewards_per_day
return int(total_staked * ratio)
if __name__ == "__main__":
config = StakingConfig(
total_staked=500_000,
reward_rate_per_block=5,
your_stake=10_000,
)
config.print_report()
print(f"\nCompounded APY (daily reinvest): {compound_apy(
config.estimated_daily_reward(),
config.your_stake,
):.2%}")
target = 50 # 目标每日 50 个代币
required = find_required_stake(
target, config.total_staked, config.reward_rate_per_block
)
print(f"\nRequired stake for {target} tokens/day: {required}")
4.4 奖励池管理
/// 奖励池
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct RewardPool {
pub reward_token: Addr, // 奖励代币地址
pub fund_balance: Uint128, // 资金余额
pub reward_balances: Map<Addr, Uint128>, // 各用户待领取
}
pub const REWARD_POOL: Item<RewardPool> = Item::new("reward_pool");
/// 向奖励池注入资金（仅管理员）
pub fn fund_reward_pool(
deps: DepsMut,
env: Env,
info: MessageInfo,
amount: Uint128,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized"));
}
// 从管理员地址转移代币到合约
cw20_base::contract::execute_transfer_from(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
env.contract.address.to_string(),
amount,
)?;
let mut pool = REWARD_POOL.load(deps.storage)?;
pool.fund_balance = pool.fund_balance.checked_add(amount)?;
REWARD_POOL.save(deps.storage, &pool)?;
Ok(Response::new()
.add_attribute("action", "fund_reward_pool")
.add_attribute("amount", amount)
.add_attribute("new_balance", pool.fund_balance))
}
5. 流动性池
5.1 流动性概述
Agent 代币需要流动性才能在 MSG Chain 上自由交易。本节展示如何在 MSG Chain 的原生 DEX 上创建和管理流动性池。
5.2 流动性池创建
#!/usr/bin/env python3
"""
流动性池管理工具
"""
import asyncio
import json
from typing import Optional, Tuple
from dataclasses import dataclass
@dataclass
class PoolInfo:
"""流动性池信息"""
pool_addr: str
token0: str # Native token denom
token1: str # CW20 token address
reserve0: int
reserve1: int
lp_token_addr: str
total_liquidity: int
class LiquidityManager:
"""流动性管理器"""
def __init__(
self,
dex_factory: str,
wallet: str,
client: "MsgChainClient",
):
self.dex_factory = dex_factory
self.wallet = wallet
self.client = client
self.gas_limit = 3000000
async def create_pool(
self,
token_addr: str,
msg_amount: int,
token_amount: int,
swap_fee: int = 30, # 0.3%
) -> Tuple[str, PoolInfo]:
"""创建 MSG-代币 流动性池
Args:
token_addr: CW20 代币地址
msg_amount: 注入的 MSG 数量（umsg）
token_amount: 注入的代币数量
swap_fee: 手续费基点 (30 = 0.3%)
Returns:
(pool_address, PoolInfo)
"""
# 1. 先批准代币转账
approve_msg = {
"contract_addr": token_addr,
"msg": {
"increase_allowance": {
"spender": self.dex_factory,
"amount": str(token_amount),
}
},
"funds": [],
}
await self.client.execute_contract_msg(
approve_msg, self.wallet, "Approve DEX"
)
# 2. 创建池
create_msg = {
"contract_addr": self.dex_factory,
"msg": {
"create_pool": {
"asset_infos": [
{"native_token": {"denom": "umsg"}},
{"token": {"contract_addr": token_addr}},
],
"initial_liquidity": [
{"amount": str(msg_amount), "denom": "umsg"},
{"amount": str(token_amount), "denom": token_addr},
],
"swap_fee": str(swap_fee),
}
},
"funds": [
{"denom": "umsg", "amount": str(msg_amount)},
],
}
tx_hash = await self.client.execute_contract_msg(
create_msg, self.wallet, "Create liquidity pool",
gas_limit=self.gas_limit,
)
# 3. 查询新池地址
pool_info = await self._query_pool(token_addr)
print(f"Pool created: {pool_info.pool_addr}")
print(f" Token0: {pool_info.token0}")
print(f" Token1: {pool_info.token1}")
print(f" Reserve: {pool_info.reserve0} umsg / {pool_info.reserve1} tokens")
return pool_info.pool_addr, pool_info
async def add_liquidity(
self,
pool_addr: str,
msg_amount: int,
token_amount: int,
slippage_tolerance: float = 0.01,
) -> dict:
"""添加流动性
Args:
pool_addr: 池子地址
msg_amount: 新增 MSG 数量
token_amount: 新增代币数量
slippage_tolerance: 滑点容忍度
Returns:
交易结果
"""
pool_info = await self._get_pool_info(pool_addr)
# 按比例计算
expected_ratio = pool_info.reserve0 / pool_info.reserve1
actual_ratio = msg_amount / token_amount
if abs(actual_ratio / expected_ratio - 1) > slippage_tolerance:
raise ValueError(
f"Price impact too high: "
f"expected ratio {expected_ratio:.4f}, "
f"actual {actual_ratio:.4f}"
)
# 批准代币
approve_msg = {
"contract_addr": pool_info.token1,
"msg": {
"increase_allowance": {
"spender": pool_addr,
"amount": str(token_amount),
}
},
"funds": [],
}
await self.client.execute_contract_msg(
approve_msg, self.wallet, "Approve LP tokens"
)
# 添加流动性
add_msg = {
"contract_addr": pool_addr,
"msg": {
"add_liquidity": {
"amounts": [
str(msg_amount),
str(token_amount),
],
"slippage_tolerance": str(slippage_tolerance),
}
},
"funds": [
{"denom": "umsg", "amount": str(msg_amount)},
],
}
tx_hash = await self.client.execute_contract_msg(
add_msg, self.wallet, "Add liquidity",
gas_limit=self.gas_limit,
)
# 查询更新后的池信息
updated_info = await self._get_pool_info(pool_addr)
result = {
"tx_hash": tx_hash,
"pool": pool_addr,
"msg_added": msg_amount,
"token_added": token_amount,
"lp_received": updated_info.total_liquidity - pool_info.total_liquidity,
"new_reserve0": updated_info.reserve0,
"new_reserve1": updated_info.reserve1,
}
print(f"Liquidity added: {result['lp_received']} LP tokens received")
return result
async def remove_liquidity(
self,
pool_addr: str,
lp_amount: int,
withdraw_percentage: float = 1.0,
) -> dict:
"""移除流动性"""
pool_info = await self._get_pool_info(pool_addr)
amount = int(lp_amount * withdraw_percentage)
remove_msg = {
"contract_addr": pool_addr,
"msg": {
"remove_liquidity": {
"amount": str(amount),
}
},
"funds": [],
}
tx_hash = await self.client.execute_contract_msg(
remove_msg, self.wallet, "Remove liquidity",
gas_limit=self.gas_limit,
)
updated_info = await self._get_pool_info(pool_addr)
removed0 = pool_info.reserve0 - updated_info.reserve0
removed1 = pool_info.reserve1 - updated_info.reserve1
return {
"tx_hash": tx_hash,
"lp_removed": amount,
"msg_received": removed0,
"tokens_received": removed1,
}
async def _query_pool(self, token_addr: str) -> PoolInfo:
"""通过代币地址查询池子"""
query = {
"pools_for_token": {
"token_addr": token_addr,
}
}
result = await self.client.query_contract(
self.dex_factory, query
)
pool_addr = result["pools"][0]
return await self._get_pool_info(pool_addr)
async def _get_pool_info(self, pool_addr: str) -> PoolInfo:
"""查询池子详情"""
query = {"pool_info": {}}
result = await self.client.query_contract(pool_addr, query)
return PoolInfo(
pool_addr=pool_addr,
token0=result["assets"][0]["info"]["native_token"]["denom"],
token1=result["assets"][1]["info"]["token"]["contract_addr"],
reserve0=int(result["assets"][0]["amount"]),
reserve1=int(result["assets"][1]["amount"]),
lp_token_addr=result["lp_token"],
total_liquidity=int(result["total_liquidity"]),
)
async def get_token_price(self, pool_addr: str) -> float:
"""获取代币价格（以 umsg 计价）"""
info = await self._get_pool_info(pool_addr)
if info.reserve1 == 0:
return 0.0
return info.reserve0 / info.reserve1
async def estimate_swap(
self,
pool_addr: str,
offer_asset: str,
offer_amount: int,
) -> int:
"""估算兑换结果
Args:
pool_addr: 池地址
offer_asset: 出售的资产（"umsg" 或代币地址）
offer_amount: 出售数量
Returns:
预期获得数量
"""
query = {
"simulate_swap": {
"offer_asset": {
"info": (
{"native_token": {"denom": offer_asset}}
if offer_asset == "umsg"
else {"token": {"contract_addr": offer_asset}}
),
"amount": str(offer_amount),
}
}
}
result = await self.client.query_contract(pool_addr, query)
return int(result["return_amount"])
async def bootstrap_liquidity(
token_addr: str,
wallet: str,
client: "MsgChainClient",
dex_factory: str,
msg_reserve: int = 100_000_000_000, # 100 MSG (in umsg)
token_reserve: int = 500_000, # 500,000 tokens
) -> PoolInfo:
"""一站式创建并初始化流动性"""
manager = LiquidityManager(dex_factory, wallet, client)
print("=" * 50)
print("Bootstrap Liquidity")
print("=" * 50)
print(f"Token: {token_addr}")
print(f"MSG Reserve: {msg_reserve} umsg")
print(f"Token Reserve: {token_reserve}")
print("\n1. Creating pool...")
pool_addr, pool_info = await manager.create_pool(
token_addr, msg_reserve, token_reserve
)
price = await manager.get_token_price(pool_addr)
print(f"\nInitial price: {price:.6f} umsg per token")
print(f"Equivalent: {1/price:,.2f} tokens per MSG")
return pool_info
if __name__ == "__main__":
import sys
TOKEN_ADDR = sys.argv[1] if len(sys.argv) > 1 else "msg1token..."
WALLET = sys.argv[2] if len(sys.argv) > 2 else "agent_admin"
# 需要实际的 client 实例
asyncio.run(bootstrap_liquidity(
token_addr=TOKEN_ADDR,
wallet=WALLET,
client=None, # 替换为实际 client
dex_factory="msg1dexfactory0000000000000000000000000",
))
5.3 LP Token 管理
class LpTokenManager:
"""LP Token 管理"""
def __init__(self, pool_addr: str, client: "MsgChainClient"):
self.pool_addr = pool_addr
self.client = client
async def get_lp_balance(self, address: str) -> int:
"""查询 LP Token 余额"""
query = {
"balance": {"address": address}
}
result = await self.client.query_contract(
self.pool_addr.replace("pool", "lp"), # LP 代币合约
query,
)
return int(result["balance"])
async def stake_lp(self, amount: int, staking_contract: str) -> str:
"""质押 LP Token 到农场"""
msg = {
"contract_addr": self.pool_addr,
"msg": {
"send": {
"contract": staking_contract,
"amount": str(amount),
"msg": b"", # 可选 hook 消息
}
},
"funds": [],
}
return await self.client.execute_contract_msg(
msg, "agent_admin", "Stake LP tokens"
)
async def get_pool_share(self, address: str) -> float:
"""查询用户占池比例"""
pool_info = await self.client.query_contract(
self.pool_addr, {"pool_info": {}}
)
total_lp = int(pool_info["total_liquidity"])
if total_lp == 0:
return 0.0
user_lp = await self.get_lp_balance(address)
return user_lp / total_lp
5.4 流动性激励
/// 流动性挖矿奖励配置
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct LiquidityMiningConfig {
pub lp_token_addr: Addr,
pub reward_token_addr: Addr,
pub reward_rate: Uint128, // 每区块奖励
pub start_block: u64,
pub end_block: u64,
pub total_rewards: Uint128,
}
pub const LIQUIDITY_MINING: Item<LiquidityMiningConfig> =
Item::new("liquidity_mining");
/// 初始化流动性挖矿
pub fn init_liquidity_mining(
deps: DepsMut,
env: Env,
info: MessageInfo,
config: LiquidityMiningConfig,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized"));
}
if config.start_block >= config.end_block {
return Err(StdError::generic_err(
"start_block must be less than end_block",
));
}
LIQUIDITY_MINING.save(deps.storage, &config)?;
Ok(Response::new()
.add_attribute("action", "init_liquidity_mining")
.add_attribute("lp_token", config.lp_token_addr)
.add_attribute("reward_token", config.reward_token_addr)
.add_attribute("total_rewards", config.total_rewards)
.add_attribute("start_block", config.start_block.to_string())
.add_attribute("end_block", config.end_block.to_string()))
}
6. 代币门控
6.1 门控机制设计
代币门控（Token Gating）是代币在 AI Agent 中的重要应用场景。Agent 可以根据用户持有的代币数量提供不同级别的服务。
访问层级:
├── 无代币: 基础功能（试用）
├── 1-100 AGT: 标准访问
├── 100-1000 AGT: 高级访问
└── 1000+ AGT: VIP 访问
6.2 代币门控合约
use cosmwasm_std::{
Addr, Binary, Deps, DepsMut, Env, MessageInfo, QueryRequest,
Response, StdError, StdResult, Uint128, WasmQuery,
};
use cw_storage_plus::Item;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
/// 门控配置
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct GatingConfig {
pub token_addr: Addr, // 门控代币地址
pub min_balance: Uint128, // 最低持有量
pub enabled: bool, // 是否启用门控
pub tiers: Vec<AccessTier>, // 多层级配置
}
/// 访问层级
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct AccessTier {
pub level: String,
pub min_tokens: Uint128,
pub features: Vec<String>,
}
pub const GATING_CONFIG: Item<GatingConfig> = Item::new("gating_config");
/// 初始化门控（仅管理员）
pub fn init_gating(
deps: DepsMut,
info: MessageInfo,
token_addr: String,
min_balance: Uint128,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized"));
}
let token = deps.api.addr_validate(&token_addr)?;
let config = GatingConfig {
token_addr: token,
min_balance,
enabled: true,
tiers: vec![
AccessTier {
level: "standard".to_string(),
min_tokens: min_balance,
features: vec![
"basic_query".to_string(),
"rate_limit_100".to_string(),
],
},
AccessTier {
level: "premium".to_string(),
min_tokens: min_balance.checked_mul(Uint128::from(10u64))?,
features: vec![
"advanced_query".to_string(),
"rate_limit_1000".to_string(),
"batch_processing".to_string(),
],
},
AccessTier {
level: "vip".to_string(),
min_tokens: min_balance.checked_mul(Uint128::from(100u64))?,
features: vec![
"all_features".to_string(),
"priority_queue".to_string(),
"dedicated_compute".to_string(),
"api_access".to_string(),
],
},
],
};
GATING_CONFIG.save(deps.storage, &config)?;
Ok(Response::new()
.add_attribute("action", "init_gating")
.add_attribute("token", token_addr)
.add_attribute("min_balance", min_balance))
}
/// 查询用户余额
pub fn query_token_balance(
deps: Deps,
token_addr: &Addr,
user_addr: &Addr,
) -> StdResult<Uint128> {
let query = Cw20QueryMsg::Balance {
address: user_addr.to_string(),
};
let balance: BalanceResponse = deps.querier.query(&QueryRequest::Wasm(
WasmQuery::Smart {
contract_addr: token_addr.to_string(),
msg: to_binary(&query)?,
},
))?;
Ok(balance.balance)
}
/// 检查用户访问权限
pub fn check_access(
deps: Deps,
user: &Addr,
) -> StdResult<String> {
let config = GATING_CONFIG.load(deps.storage)?;
if !config.enabled {
return Ok("unrestricted".to_string());
}
let balance = query_token_balance(
deps,
&config.token_addr,
user,
)?;
if balance < config.min_balance {
return Err(StdError::generic_err(format!(
"Insufficient tokens. Required: {}, Balance: {}",
config.min_balance, balance,
)));
}
// 确定层级
for tier in config.tiers.iter().rev() {
if balance >= tier.min_tokens {
return Ok(tier.level.clone());
}
}
Ok("standard".to_string())
}
/// 获取用户可访问的功能
pub fn get_accessible_features(
deps: Deps,
user: &Addr,
) -> StdResult<Vec<String>> {
let config = GATING_CONFIG.load(deps.storage)?;
let balance = query_token_balance(deps, &config.token_addr, user)?;
let mut features = vec![];
for tier in &config.tiers {
if balance >= tier.min_tokens {
features.extend(tier.features.clone());
}
}
Ok(features)
}
/// 基于代币的速率限制
pub fn check_rate_limit(
deps: Deps,
user: &Addr,
usage_count: u64,
) -> StdResult<bool> {
let config = GATING_CONFIG.load(deps.storage)?;
let balance = query_token_balance(deps, &config.token_addr, user)?;
// 根据余额确定速率限制
let limit = if balance >= config.min_balance * Uint128::from(100u64) {
1000 // VIP: 1000 次/分钟
} else if balance >= config.min_balance * Uint128::from(10u64) {
100 // Premium: 100 次/分钟
} else if balance >= config.min_balance {
10 // Standard: 10 次/分钟
} else {
1 // 无权限
};
Ok(usage_count < limit)
}
6.3 基于订阅的代币模型
/// 订阅计划
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct SubscriptionPlan {
pub name: String,
pub description: String,
pub token_cost: Uint128, // 每月代币费用
pub duration_seconds: u64, // 订阅时长（秒）
pub features: Vec<String>,
}
pub const SUBSCRIPTION_PLANS: Map<&str, SubscriptionPlan> =
Map::new("subscription_plans");
/// 用户订阅
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct UserSubscription {
pub user: Addr,
pub plan: String,
pub start_time: u64,
pub end_time: u64,
pub auto_renew: bool,
}
pub const USER_SUBSCRIPTIONS: Map<&Addr, UserSubscription> =
Map::new("user_subscriptions");
/// 订阅代币服务
pub fn subscribe(
deps: DepsMut,
env: Env,
info: MessageInfo,
plan_name: String,
auto_renew: bool,
) -> StdResult<Response> {
let plan = SUBSCRIPTION_PLANS
.load(deps.storage, &plan_name)?;
// 检查用户余额
let balance = query_token_balance(
deps.as_ref(),
&GATING_CONFIG.load(deps.storage)?.token_addr,
&info.sender,
)?;
if balance < plan.token_cost {
return Err(StdError::generic_err(format!(
"Insufficient balance. Need {} tokens, have {}",
plan.token_cost, balance,
)));
}
// 转移代币
cw20_base::contract::execute_transfer_from(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
env.contract.address.to_string(),
plan.token_cost,
)?;
// 创建订阅
let subscription = UserSubscription {
user: info.sender.clone(),
plan: plan_name.clone(),
start_time: env.block.time.seconds(),
end_time: env.block.time.seconds() + plan.duration_seconds,
auto_renew,
};
USER_SUBSCRIPTIONS.save(
deps.storage,
&info.sender,
&subscription,
)?;
Ok(Response::new()
.add_attribute("action", "subscribe")
.add_attribute("user", info.sender)
.add_attribute("plan", plan_name)
.add_attribute("cost", plan.token_cost)
.add_attribute("end_time", subscription.end_time.to_string()))
}
/// 检查订阅状态
pub fn check_subscription(
deps: Deps,
user: &Addr,
) -> StdResult<Option<UserSubscription>> {
let sub = USER_SUBSCRIPTIONS.may_load(deps.storage, user)?;
Ok(sub)
}
/// 续订
pub fn renew_subscription(
deps: DepsMut,
env: Env,
info: MessageInfo,
) -> StdResult<Response> {
let mut sub = USER_SUBSCRIPTIONS
.load(deps.storage, &info.sender)?;
if sub.end_time > env.block.time.seconds() {
return Err(StdError::generic_err(
"Subscription still active",
));
}
let plan = SUBSCRIPTION_PLANS
.load(deps.storage, &sub.plan)?;
// 扣除代币
cw20_base::contract::execute_transfer_from(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
env.contract.address.to_string(),
plan.token_cost,
)?;
sub.start_time = env.block.time.seconds();
sub.end_time = env.block.time.seconds() + plan.duration_seconds;
USER_SUBSCRIPTIONS.save(deps.storage, &info.sender, &sub)?;
Ok(Response::new()
.add_attribute("action", "renew")
.add_attribute("user", info.sender)
.add_attribute("plan", sub.plan)
.add_attribute("end_time", sub.end_time.to_string()))
}
6.4 门控中间件示例
#!/usr/bin/env python3
"""
代币门控中间件
用于保护 Agent 的 API 端点
"""
import time
from functools import wraps
from typing import Optional, Callable
class TokenGateMiddleware:
"""API 代币门控中间件"""
def __init__(
self,
token_addr: str,
client: "MsgChainClient",
rate_limit_enabled: bool = True,
):
self.token_addr = token_addr
self.client = client
self.rate_limit_enabled = rate_limit_enabled
self._usage: dict = {}
self._tier_cache: dict = {}
self._cache_ttl = 60 # 缓存 60 秒
async def get_user_tier(self, user_addr: str) -> str:
"""查询用户层级（带缓存）"""
now = time.time()
cached = self._tier_cache.get(user_addr)
if cached and now - cached["timestamp"] < self._cache_ttl:
return cached["tier"]
try:
query = {
"check_access": {
"user": user_addr,
}
}
result = await self.client.query_contract(
self.token_addr, query
)
tier = result.get("level", "none")
except Exception:
tier = "none"
self._tier_cache[user_addr] = {
"tier": tier,
"timestamp": now,
}
return tier
def check_rate_limit(self, user_addr: str) -> bool:
"""检查速率限制"""
if not self.rate_limit_enabled:
return True
now = int(time.time())
minute_key = now // 60
# 清理旧数据
self._usage = {
k: v for k, v in self._usage.items()
if k[1] >= minute_key - 1
}
key = (user_addr, minute_key)
count = self._usage.get(key, 0) + 1
self._usage[key] = count
# 根据不同层级设置限制
tier_limits = {
"none": 0,
"standard": 10,
"premium": 100,
"vip": 1000,
"unrestricted": 999999,
}
return count <= tier_limits.get(
self._tier_cache.get(user_addr, {}).get("tier", "none"),
10,
)
def require_tier(
self,
min_tier: str = "standard",
) -> Callable:
"""装饰器：要求最低访问层级"""
tier_order = ["none", "standard", "premium", "vip", "unrestricted"]
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(user_addr: str, *args, **kwargs):
tier = await self.get_user_tier(user_addr)
if tier_order.index(tier) < tier_order.index(min_tier):
raise PermissionError(
f"Insufficient access tier. "
f"Required: {min_tier}, Current: {tier}"
)
if not self.check_rate_limit(user_addr):
raise PermissionError(
"Rate limit exceeded"
)
return await func(user_addr, *args, **kwargs)
return wrapper
return decorator
# ===== 使用示例 =====
gate = TokenGateMiddleware(
token_addr="msg1token...",
client=client_instance,
)
@gate.require_tier("standard")
async def basic_query(user_addr: str, question: str) -> str:
"""基础查询 - 需要 Standard 以上"""
return f"Answer to: {question}"
@gate.require_tier("premium")
async def advanced_query(user_addr: str, prompt: str) -> str:
"""高级查询 - 需要 Premium 以上"""
return f"Advanced analysis: {prompt}"
@gate.require_tier("vip")
async def dedicated_inference(
user_addr: str,
model: str,
params: dict,
) -> bytes:
"""专属推理 - 需要 VIP"""
return b"model output"
7. 治理代币
7.1 治理机制设计
治理代币赋予持有者对 Agent 的决策权。核心功能包括提案创建、投票和委托。
7.2 治理合约
use cosmwasm_std::{
Addr, Decimal, Deps, DepsMut, Env, MessageInfo,
Response, StdError, StdResult, Storage, Uint128,
};
use cw_storage_plus::{Item, Map};
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
// ===== 类型定义 =====
#[derive(Serialize, Deserialize, Clone, Debug)]
pub enum VoteOption {
Yes,
No,
Abstain,
Veto,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Proposal {
pub id: u64,
pub title: String,
pub description: String,
pub proposer: Addr,
pub start_time: u64,
pub end_time: u64,
pub status: ProposalStatus,
pub yes_votes: Uint128,
pub no_votes: Uint128,
pub abstain_votes: Uint128,
pub veto_votes: Uint128,
pub total_voting_power: Uint128,
pub executed: bool,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq)]
pub enum ProposalStatus {
Active,
Passed,
Rejected,
Vetoed,
Executed,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct Vote {
pub voter: Addr,
pub proposal_id: u64,
pub option: VoteOption,
pub weight: Uint128,
pub timestamp: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug)]
pub struct GovernanceConfig {
pub token_addr: Addr,
pub min_proposal_deposit: Uint128,
pub quorum: Decimal, // 法定投票率
pub threshold: Decimal, // 通过门槛
pub veto_threshold: Decimal, // 否决门槛
pub voting_period: u64, // 投票周期（秒）
pub proposal_deposit: Uint128,
}
// ===== 存储 =====
pub const GOV_CONFIG: Item<GovernanceConfig> = Item::new("gov_config");
pub const PROPOSALS: Map<u64, Proposal> = Map::new("proposals");
pub const VOTES: Map<(u64, &Addr), Vote> = Map::new("votes");
pub const DELEGATIONS: Map<&Addr, Addr> = Map::new("delegations");
pub const NEXT_PROPOSAL_ID: Item<u64> = Item::new("next_proposal_id");
// ===== 初始化 =====
pub fn init_governance(
deps: DepsMut,
info: MessageInfo,
token_addr: String,
quorum: Decimal,
threshold: Decimal,
voting_period: u64,
proposal_deposit: Uint128,
) -> StdResult<Response> {
let metadata = TOKEN_METADATA.load(deps.storage)?;
if info.sender != metadata.admin {
return Err(StdError::generic_err("Unauthorized"));
}
let config = GovernanceConfig {
token_addr: deps.api.addr_validate(&token_addr)?,
min_proposal_deposit: proposal_deposit,
quorum,
threshold,
veto_threshold: Decimal::percent(33), // 33% veto
voting_period,
proposal_deposit,
};
GOV_CONFIG.save(deps.storage, &config)?;
NEXT_PROPOSAL_ID.save(deps.storage, &1)?;
Ok(Response::new()
.add_attribute("action", "init_governance")
.add_attribute("token", token_addr))
}
// ===== 提案 =====
pub fn submit_proposal(
deps: DepsMut,
env: Env,
info: MessageInfo,
title: String,
description: String,
) -> StdResult<Response> {
let config = GOV_CONFIG.load(deps.storage)?;
if title.is_empty() || title.len() > 200 {
return Err(StdError::generic_err(
"Title must be 1-200 characters",
));
}
if description.is_empty() || description.len() > 10000 {
return Err(StdError::generic_err(
"Description must be 1-10000 characters",
));
}
// 检查余额并锁定押金
let balance = query_token_balance(
deps.as_ref(),
&config.token_addr,
&info.sender,
)?;
if balance < config.proposal_deposit {
return Err(StdError::generic_err(format!(
"Need {} tokens for proposal deposit, have {}",
config.proposal_deposit, balance,
)));
}
// 转移押金到合约
cw20_base::contract::execute_transfer_from(
deps.branch(),
env.clone(),
info.clone(),
info.sender.to_string(),
env.contract.address.to_string(),
config.proposal_deposit,
)?;
let id = NEXT_PROPOSAL_ID.load(deps.storage)?;
NEXT_PROPOSAL_ID.save(deps.storage, &(id + 1))?;
let proposal = Proposal {
id,
title,
description,
proposer: info.sender,
start_time: env.block.time.seconds(),
end_time: env.block.time.seconds() + config.voting_period,
status: ProposalStatus::Active,
yes_votes: Uint128::zero(),
no_votes: Uint128::zero(),
abstain_votes: Uint128::zero(),
veto_votes: Uint128::zero(),
total_voting_power: Uint128::zero(),
executed: false,
};
PROPOSALS.save(deps.storage, &id, &proposal)?;
Ok(Response::new()
.add_attribute("action", "submit_proposal")
.add_attribute("proposal_id", id.to_string())
.add_attribute("title", proposal.title))
}
// ===== 投票 =====
pub fn cast_vote(
deps: DepsMut,
env: Env,
info: MessageInfo,
proposal_id: u64,
option: VoteOption,
) -> StdResult<Response> {
let config = GOV_CONFIG.load(deps.storage)?;
let mut proposal = PROPOSALS.load(deps.storage, &proposal_id)?;
// 检查投票是否已结束
let now = env.block.time.seconds();
if now > proposal.end_time {
return Err(StdError::generic_err("Voting period has ended"));
}
if proposal.status != ProposalStatus::Active {
return Err(StdError::generic_err("Proposal is not active"));
}
// 检查是否已投票
if VOTES.has(deps.storage, &(proposal_id, &info.sender)) {
return Err(StdError::generic_err("Already voted"));
}
// 检查投票人是否已投票
if VOTES.has(deps.storage, &(proposal_id, &info.sender)) {
return Err(StdError::generic_err("Duplicate vote"));
}
// 获取投票权重
let voting_power = get_voting_power(deps.as_ref(), &config, &info.sender)?;
if voting_power.is_zero() {
return Err(StdError::generic_err("Zero voting power"));
}
// 记录投票
let vote = Vote {
voter: info.sender.clone(),
proposal_id,
option: option.clone(),
weight: voting_power,
timestamp: now,
};
VOTES.save(deps.storage, &(proposal_id, &info.sender), &vote)?;
// 更新提案统计
match option {
VoteOption::Yes => {
proposal.yes_votes =
proposal.yes_votes.checked_add(voting_power)?;
}
VoteOption::No => {
proposal.no_votes =
proposal.no_votes.checked_add(voting_power)?;
}
VoteOption::Abstain => {
proposal.abstain_votes =
proposal.abstain_votes.checked_add(voting_power)?;
}
VoteOption::Veto => {
proposal.veto_votes =
proposal.veto_votes.checked_add(voting_power)?;
}
}
proposal.total_voting_power =
proposal.total_voting_power.checked_add(voting_power)?;
PROPOSALS.save(deps.storage, &proposal_id, &proposal)?;
Ok(Response::new()
.add_attribute("action", "vote")
.add_attribute("proposal_id", proposal_id.to_string())
.add_attribute("voter", info.sender)
.add_attribute("option", format!("{:?}", option))
.add_attribute("weight", voting_power))
}
/// 获取投票权重
pub fn get_voting_power(
deps: Deps,
config: &GovernanceConfig,
voter: &Addr,
) -> StdResult<Uint128> {
// 检查委托
let effective_voter = DELEGATIONS
.may_load(deps.storage, voter)?
.unwrap_or_else(|| voter.clone());
// 查询代币余额
query_token_balance(deps, &config.token_addr, &effective_voter)
}
// ===== 委托 =====
pub fn delegate(
deps: DepsMut,
_env: Env,
info: MessageInfo,
delegate_to: String,
) -> StdResult<Response> {
let delegate_addr = deps.api.addr_validate(&delegate_to)?;
if info.sender == delegate_addr {
return Err(StdError::generic_err(
"Cannot delegate to self",
));
}
DELEGATIONS.save(deps.storage, &info.sender, &delegate_addr)?;
Ok(Response::new()
.add_attribute("action", "delegate")
.add_attribute("delegator", info.sender)
.add_attribute("delegate_to", delegate_to))
}
pub fn undelegate(
deps: DepsMut,
_env: Env,
info: MessageInfo,
) -> StdResult<Response> {
DELEGATIONS.remove(deps.storage, &info.sender);
Ok(Response::new()
.add_attribute("action", "undelegate")
.add_attribute("delegator", info.sender))
}
// ===== 提案执行 =====
pub fn execute_proposal(
deps: DepsMut,
env: Env,
info: MessageInfo,
proposal_id: u64,
) -> StdResult<Response> {
let config = GOV_CONFIG.load(deps.storage)?;
let mut proposal = PROPOSALS.load(deps.storage, &proposal_id)?;
let now = env.block.time.seconds();
if now < proposal.end_time {
return Err(StdError::generic_err(
"Voting period not ended",
));
}
if proposal.executed {
return Err(StdError::generic_err(
"Proposal already executed",
));
}
// 计算结果
let total_votes = proposal.yes_votes
+ proposal.no_votes
+ proposal.abstain_votes
+ proposal.veto_votes;
// 获取总投票权
let total_supply = query_token_supply(deps.as_ref(), &config.token_addr)?;
// 检查法定人数
let turnout = if total_supply.is_zero() {
Decimal::zero()
} else {
Decimal::from_ratio(total_votes, total_supply)
};
if turnout < config.quorum {
proposal.status = ProposalStatus::Rejected;
PROPOSALS.save(deps.storage, &proposal_id, &proposal)?;
return Err(StdError::generic_err(format!(
"Quorum not met: {} < {}",
turnout, config.quorum,
)));
}
// 检查否决
let veto_ratio = if total_votes.is_zero() {
Decimal::zero()
} else {
Decimal::from_ratio(proposal.veto_votes, total_votes)
};
if veto_ratio > config.veto_threshold {
proposal.status = ProposalStatus::Vetoed;
PROPOSALS.save(deps.storage, &proposal_id, &proposal)?;
return Err(StdError::generic_err("Proposal vetoed"));
}
// 检查通过
let yes_ratio = if total_votes.is_zero() {
Decimal::zero()
} else {
Decimal::from_ratio(proposal.yes_votes, total_votes)
};
if yes_ratio > config.threshold {
proposal.status = ProposalStatus::Passed;
proposal.executed = true;
PROPOSALS.save(deps.storage, &proposal_id, &proposal)?;
// 退还押金
cw20_base::contract::execute_transfer(
deps.branch(),
env.clone(),
info.clone(),
proposal.proposer.to_string(),
config.proposal_deposit,
)?;
Ok(Response::new()
.add_attribute("action", "execute_proposal")
.add_attribute("proposal_id", proposal_id.to_string())
.add_attribute("result", "passed"))
} else {
proposal.status = ProposalStatus::Rejected;
PROPOSALS.save(deps.storage, &proposal_id, &proposal)?;
Err(StdError::generic_err("Proposal rejected"))
}
}
7.3 治理查询
/// 查询所有提案
pub fn list_proposals(
deps: Deps,
start_after: Option<u64>,
limit: Option<u32>,
) -> StdResult<Binary> {
let limit = limit.unwrap_or(30).min(100);
let start = start_after.unwrap_or(0);
let proposals: Vec<Proposal> = PROPOSALS
.range(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.filter(|r| r.as_ref().map(|(id, _)| *id > start).unwrap_or(false))
.take(limit as usize)
.map(|r| r.map(|(_, p)| p))
.collect::<StdResult<Vec<_>>>()?;
to_binary(&proposals)
}
/// 查询单个提案
pub fn query_proposal(deps: Deps, proposal_id: u64) -> StdResult<Binary> {
let proposal = PROPOSALS.load(deps.storage, &proposal_id)?;
to_binary(&proposal)
}
/// 查询投票记录
pub fn query_votes(
deps: Deps,
proposal_id: u64,
) -> StdResult<Binary> {
let votes: Vec<Vote> = VOTES
.prefix(proposal_id)
.range(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.map(|r| r.map(|(_, v)| v))
.collect::<StdResult<Vec<_>>>()?;
to_binary(&votes)
}
/// 查询委托人
pub fn query_delegation(deps: Deps, delegator: String) -> StdResult<Binary> {
let addr = deps.api.addr_validate(&delegator)?;
let delegate = DELEGATIONS.may_load(deps.storage, &addr)?;
to_binary(&delegate)
}
7.4 治理 Python 客户端
#!/usr/bin/env python3
"""
治理客户端
"""
from dataclasses import dataclass
from typing import Optional, List
from enum import Enum
class VoteOption(Enum):
YES = "yes"
NO = "no"
ABSTAIN = "abstain"
VETO = "veto"
@dataclass
class ProposalInfo:
id: int
title: str
description: str
proposer: str
status: str
yes_votes: int
no_votes: int
abstain_votes: int
veto_votes: int
end_time: int
class GovernanceClient:
"""治理客户端"""
def __init__(self, gov_addr: str, wallet: str, client):
self.gov_addr = gov_addr
self.wallet = wallet
self.client = client
async def submit_proposal(
self,
title: str,
description: str,
) -> int:
"""提交提案"""
msg = {
"submit_proposal": {
"title": title,
"description": description,
}
}
result = await self.client.execute_contract(
self.gov_addr, msg, self.wallet
)
for attr in result["logs"][0]["events"][0]["attributes"]:
if attr["key"] == "proposal_id":
return int(attr["value"])
raise RuntimeError("Could not get proposal ID")
async def vote(
self,
proposal_id: int,
option: VoteOption,
) -> str:
"""投票"""
msg = {
"cast_vote": {
"proposal_id": proposal_id,
"option": option.value,
}
}
return await self.client.execute_contract(
self.gov_addr, msg, self.wallet
)
async def delegate(self, delegate_to: str) -> str:
"""委托投票权"""
msg = {
"delegate": {
"delegate_to": delegate_to,
}
}
return await self.client.execute_contract(
self.gov_addr, msg, self.wallet
)
async def get_proposal(self, proposal_id: int) -> ProposalInfo:
"""查询提案"""
query = {"query_proposal": {"proposal_id": proposal_id}}
result = await self.client.query_contract(self.gov_addr, query)
return ProposalInfo(
id=result["id"],
title=result["title"],
description=result["description"],
proposer=result["proposer"],
status=result["status"],
yes_votes=int(result["yes_votes"]),
no_votes=int(result["no_votes"]),
abstain_votes=int(result["abstain_votes"]),
veto_votes=int(result["veto_votes"]),
end_time=result["end_time"],
)
async def list_proposals(
self,
start_after: Optional[int] = None,
limit: int = 30,
) -> List[ProposalInfo]:
"""列出提案"""
query = {
"list_proposals": {
"start_after": start_after,
"limit": limit,
}
}
results = await self.client.query_contract(self.gov_addr, query)
return [
ProposalInfo(
id=p["id"],
title=p["title"],
description=p["description"],
proposer=p["proposer"],
status=p["status"],
yes_votes=int(p["yes_votes"]),
no_votes=int(p["no_votes"]),
abstain_votes=int(p["abstain_votes"]),
veto_votes=int(p["veto_votes"]),
end_time=p["end_time"],
)
for p in results
]
async def get_proposal_results(self, proposal_id: int) -> dict:
"""获取提案结果"""
proposal = await self.get_proposal(proposal_id)
total = (
proposal.yes_votes
+ proposal.no_votes
+ proposal.abstain_votes
+ proposal.veto_votes
)
return {
"proposal_id": proposal.id,
"status": proposal.status,
"total_votes": total,
"yes": proposal.yes_votes,
"no": proposal.no_votes,
"abstain": proposal.abstain_votes,
"veto": proposal.veto_votes,
"yes_pct": proposal.yes_votes / total * 100 if total else 0,
"no_pct": proposal.no_votes / total * 100 if total else 0,
}
8. 完整示例：Agent Token 启动
8.1 一键启动脚本
#!/usr/bin/env python3
"""
一站式 Agent 代币启动脚本
包含：合约部署、分发、质押池、流动性池、门控配置
用法:
python3 launch_agent_token.py \
--agent-did "did:msg:agent:my-ai-agent-v1" \
--token-name "My Agent Token" \
--symbol "MAT" \
--supply 10000000 \
--admin msg1admin...
"""
import json
import sys
import asyncio
import argparse
from pathlib import Path
from typing import Optional
async def launch_agent_token(
agent_did: str,
token_name: str,
symbol: str,
total_supply: int,
admin_addr: str,
wallet: str = "agent_admin",
contract_path: str = "./agent-token",
dex_factory: str = "msg1dexfactory0000000000000000000000000",
airdrop_file: Optional[str] = None,
team_file: Optional[str] = None,
):
"""完成 Agent 代币发布的所有步骤"""
steps = [
("编译合约", False),
("部署合约", False),
("创建分发计划", False),
("执行空投", airdrop_file is not None),
("设置团队归属", team_file is not None),
("创建质押池", False),
("添加流动性", False),
("配置代币门控", False),
("初始化治理", False),
]
completed = []
for step_name, conditional in steps:
if not conditional:
completed.append(step_name)
print("=" * 60)
print(f"Launching {token_name} ({symbol})")
print(f"Agent DID: {agent_did}")
print(f"Supply: {total_supply:,}")
print(f"Admin: {admin_addr}")
print("=" * 60)
# Step 1: 编译
print("\n[1/9] Compiling smart contract...")
# 实际调用 docker 编译
wasm_path = compile_contract(contract_path)
print(f" Compiled: {wasm_path}")
# Step 2: 部署
print("\n[2/9] Deploying to MSG Chain...")
code_id = store_code(
wasm_path, wallet,
label=f"{symbol}-agent-token",
)
print(f" Code ID: {code_id}")
# 初始化代币配置
token_config = {
"name": token_name,
"symbol": symbol,
"decimals": 6,
"initial_balances": [
{"address": admin_addr, "amount": str(total_supply)}
],
"mint": {
"minter": admin_addr,
"cap": None,
},
"agent_did": agent_did,
"token_type": "utility",
"admin": admin_addr,
}
contract_addr = instantiate_contract(
code_id, token_config, wallet,
admin=admin_addr,
)
print(f" Contract: {contract_addr}")
# Step 3: 分发计划
print("\n[3/9] Creating distribution plan...")
config = DistributionConfig(total_supply=total_supply)
print(f" Community Sale: {config.community_sale_amount:,}")
print(f" Ecosystem Fund: {config.ecosystem_fund_amount:,}")
print(f" Team Vesting: {config.team_vesting_amount:,}")
print(f" Liquidity: {config.liquidity_amount:,}")
print(f" Airdrop: {config.airdrop_amount:,}")
# Step 4: 空投
if airdrop_file:
print("\n[4/9] Executing airdrop...")
distributor = TokenDistributor(contract_addr, wallet, client)
tx_hashes = await distributor.airdrop_from_csv(airdrop_file)
print(f" Airdrop complete: {len(tx_hashes)} transactions")
else:
print("\n[4/9] Skipping airdrop (no file provided)")
# Step 5: 团队归属
if team_file:
print("\n[5/9] Setting up team vesting...")
with open(team_file) as f:
team_data = json.load(f)
total_weight = sum(m["weight"] for m in team_data["members"])
schedules = []
for member in team_data["members"]:
allocation = int(
config.team_vesting_amount
* member["weight"]
/ total_weight
)
schedules.append({
"address": member["address"],
"amount": allocation,
})
print(f" {member['name']}: {allocation:,} tokens")
else:
print("\n[5/9] Skipping team vesting (no file provided)")
# Step 6: 质押池
print("\n[6/9] Creating staking pool...")
reward_rate = total_supply // (4 * 5256000) # 4年线性释放
print(f" Reward rate: {reward_rate} tokens/block")
print(f" Estimated APY at 50% staking: {reward_rate * 5256000 / (total_supply // 2):.2%}")
# Step 7: 流动性
print("\n[7/9] Adding liquidity...")
msg_liquidity = config.liquidity_amount // 1000 # 假设价格 1000 umsg/token
token_liquidity = config.liquidity_amount
print(f" MSG: {msg_liquidity:,} umsg")
print(f" Tokens: {token_liquidity:,}")
# Step 8: 门控
print("\n[8/9] Configuring token gating...")
tiers = [
("Standard", total_supply // 10000),
("Premium", total_supply // 1000),
("VIP", total_supply // 100),
]
for name, min_tokens in tiers:
print(f" {name}: {min_tokens:,} tokens")
# Step 9: 治理
print("\n[9/9] Initializing governance...")
print(f" Quorum: 20%")
print(f" Threshold: 50%")
print(f" Voting period: 3 days")
print(f" Proposal deposit: {total_supply // 10000:,} tokens")
# 输出汇总
print("\n" + "=" * 60)
print("LAUNCH COMPLETE")
print("=" * 60)
print(f"Token: {token_name} ({symbol})")
print(f"Contract: {contract_addr}")
print(f"Code ID: {code_id}")
print(f"Agent: {agent_did}")
print(f"Explorer: https://msgscan.ai/tokens/{contract_addr}")
print("=" * 60)
return {
"token_addr": contract_addr,
"code_id": code_id,
"symbol": symbol,
"name": token_name,
}
def main():
parser = argparse.ArgumentParser(
description="Launch Agent Token on MSG Chain"
)
parser.add_argument("--agent-did", required=True)
parser.add_argument("--token-name", required=True)
parser.add_argument("--symbol", required=True)
parser.add_argument("--supply", type=int, default=10_000_000)
parser.add_argument("--admin", required=True)
parser.add_argument("--wallet", default="agent_admin")
parser.add_argument("--airdrop-csv")
parser.add_argument("--team-json")
args = parser.parse_args()
asyncio.run(launch_agent_token(
agent_did=args.agent_did,
token_name=args.token_name,
symbol=args.symbol,
total_supply=args.supply,
admin_addr=args.admin,
wallet=args.wallet,
airdrop_file=args.airdrop_csv,
team_file=args.team_json,
))
if __name__ == "__main__":
main()
8.2 完整测试套件
#[cfg(test)]
mod integration_tests {
use super::*;
use cosmwasm_std::testing::{
mock_dependencies, mock_env, mock_info,
};
use cosmwasm_std::{
coin, coins, from_binary, Addr, Uint128,
};
const ADMIN: &str = "msg1admin00000000000000000000000000000000";
const USER: &str = "msg1user0000000000000000000000000000000000";
const AGENT_DID: &str = "did:msg:agent:integration-test";
fn setup_full_contract(deps: DepsMut) -> Addr {
let env = mock_env();
let info = mock_info(ADMIN, &[]);
// 部署代币
let init_msg = InstantiateMsg {
name: "Integration Token".to_string(),
symbol: "INT".to_string(),
decimals: 6,
initial_balances: vec![Cw20Coin {
address: ADMIN.to_string(),
amount: Uint128::new(1_000_000_000_000),
}],
mint: Some(MinterResponse {
minter: ADMIN.to_string(),
cap: None,
}),
agent_did: AGENT_DID.to_string(),
token_type: "governance".to_string(),
admin: ADMIN.to_string(),
};
let contract_addr = env.contract.address.clone();
instantiate(deps, env, info, init_msg).unwrap();
contract_addr
}
#[test]
fn full_token_lifecycle() {
let mut deps = mock_dependencies();
let _contract = setup_full_contract(deps.as_mut());
let env = mock_env();
// 1. 转账
let info = mock_info(ADMIN, &[]);
execute(
deps.as_mut(),
env.clone(),
info,
ExecuteMsg::Transfer {
recipient: USER.to_string(),
amount: Uint128::new(1000),
},
)
.unwrap();
// 2. 铸造
let info = mock_info(ADMIN, &[]);
execute(
deps.as_mut(),
env.clone(),
info,
ExecuteMsg::Mint {
recipient: ADMIN.to_string(),
amount: Uint128::new(5000),
},
)
.unwrap();
// 3. 销毁
let info = mock_info(ADMIN, &[]);
execute(
deps.as_mut(),
env.clone(),
info,
ExecuteMsg::Burn {
amount: Uint128::new(100),
},
)
.unwrap();
// 4. 质押
let info = mock_info(USER, &[]);
// 授权合约转移
cw20_base::contract::execute_increase_allowance(
deps.as_mut().branch(),
env.clone(),
info.clone(),
env.contract.address.to_string(),
Uint128::new(500),
)
.unwrap();
// 5. 查询最终余额
let res = query(
deps.as_ref(),
env.clone(),
QueryMsg::Balance {
address: ADMIN.to_string(),
},
)
.unwrap();
let balance: BalanceResponse = from_binary(&res).unwrap();
assert_eq!(
balance.balance,
Uint128::new(1_000_000_000_000 + 5000 - 1000 - 100)
);
// 6. 查询元数据
let res = query(
deps.as_ref(),
env,
QueryMsg::AgentMetadata {},
)
.unwrap();
let metadata: TokenMetadata = from_binary(&res).unwrap();
assert_eq!(metadata.agent_did, AGENT_DID);
assert_eq!(metadata.token_type, "governance");
}
#[test]
fn non_admin_cannot_pause() {
let mut deps = mock_dependencies();
setup_full_contract(deps.as_mut());
let env = mock_env();
let info = mock_info(USER, &[]);
let err = execute(
deps.as_mut(),
env,
info,
ExecuteMsg::Pause {},
)
.unwrap_err();
assert_eq!(err.to_string(), "Unauthorized: admin only");
}
}
8.3 安全建议
#!/usr/bin/env python3
"""
代币安全审计清单
"""
SECURITY_CHECKLIST = {
"合约安全": [
"是否使用经过审计的 cw20-base 作为基础？",
"Mint 权限是否只绑定到管理员地址？",
"是否实现了暂停/恢复机制？",
"批量操作是否有上限防止 gas 耗尽？",
"Vesting 的悬崖期和线性解锁是否正确实现？",
],
"经济安全": [
"代币总量是否有上限？",
"团队解锁是否有足够的悬崖期（>1年）？",
"质押奖励率是否可持续？",
"流动性池是否有足够的深度？",
"空投机制是否防女巫攻击？",
],
"治理安全": [
"提案押金是否足够防止垃圾提案？",
"法定人数（Quorum）是否合理？",
"是否有否决机制保护少数？",
"委托投票是否正确计算权重？",
"提案执行是否有时间锁？",
],
"操作安全": [
"管理员密钥是否使用多签？",
"合约是否有升级机制？",
"是否有完善的监控和告警？",
"紧急暂停是否可用？",
"是否有定期安全审计？",
],
}
def print_security_report():
"""输出安全审计报告"""
print("=" * 60)
print("Agent Token Security Checklist")
print("=" * 60)
for category, items in SECURITY_CHECKLIST.items():
print(f"\n[{category}]")
for i, item in enumerate(items, 1):
print(f" {i}. [ ] {item}")
print("\n" + "=" * 60)
print("Review each item before mainnet launch!")
print("=" * 60)
附录
A. MSG Chain 常用合约地址
| 合约 | 地址 |
|---|---|
| DEX 工厂 | msg1...factory |
| 质押模板 | msg1...staking |
| 治理模板 | msg1...gov |
| 门控模板 | msg1...gate |
B. 常用 CLI 命令
# 查询代币信息
msgcli query wasm contract-state smart <token_addr> '{"token_info":{}}'
# 查询余额
msgcli query wasm contract-state smart <token_addr> \
'{"balance":{"address":"msg1..."}}'
# 转账
msgcli tx wasm execute <token_addr> \
'{"transfer":{"recipient":"msg1...","amount":"1000000"}}' \
--from wallet_name
# 查询交易
msgcli query tx <tx_hash>
C. 代币经济学模板
代币名称: [Agent Token Name]
代币符号: [SYMBOL]
总供应量: [10,000,000]
精度: 6 (注: 这是 CW20 代币模板示例，原生 MSG 代币精度为 18 位小数)
分发方案:
社区销售: 30% (3,000,000) — 初始价格 [X] umsg
生态基金: 25% (2,500,000) — DAO 管理
团队归属: 20% (2,000,000) — 4年解锁, 1年悬崖
流动性: 15% (1,500,000) — DEX 初始流动性
空投: 10% (1,000,000) — 早期贡献者
质押:
奖励率: [X] tokens/block
最低质押: [100] tokens
解绑周期: [14] 天
治理:
法定人数: 20%
通过门槛: 50%
否决门槛: 33%
投票周期: 3 天
提案押金: [10,000] tokens
D. 常见问题
Q: 部署合约需要多少 MSG？
A: 存储合约约需 0.5-1 MSG，实例化约需 0.1-0.3 MSG。建议准备至少 5 MSG。
Q: 如何更新代币元数据？
A: 管理员可以调用 UpdateMetadata 更新 DID 和管理员地址。
Q: 代币可以增发吗？
A: 部署时指定了 Minter 地址即可铸造新代币。建议在合约中设置供应上限。
Q: 如何停止合约？
A: 管理员可以调用 Pause 暂停所有操作。暂停后只有 Unpause 可用。
Q: 代币精度用多少合适？
A: 推荐 6 位（同 MSG Chain 原生代币），也可使用 18 位（以太坊兼容）。
链: msg-chain-1 | 前缀: msg