MSG Chain AI Agent NFT 铸造与数字权益指南
链 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 需要 NFT
AI Agent 在 MSG Chain 上运行，天然需要去中心化身份和可验证的链上凭证。NFT（非同质化代币）为 Agent 提供了以下核心能力：
| 用途 | 说明 | 典型场景 |
|---|---|---|
| 成就证明 | 将 Agent 的里程碑铸造为 NFT | 完成 10,000 次交易、达到 95% 准确率 |
| 信誉标识 | 可验证的信誉评分 NFT | Reputation Badge Level 5 |
| 访问权限 | NFT 门控高级功能 | 持有会员通行证才能调用 premium API |
| 身份凭证 | 链上身份标识 | Agent DID 绑定 NFT |
| 贡献记录 | 记录 Agent 对生态的贡献 | 代码提交、治理投票、数据贡献 |
| 会员资格 | 时间限制的会员 NFT | 30 天高级会员、年度白金会员 |
NFT 对比传统中心化凭证的优势：
- 不可篡改：链上存储，任何方无法伪造
- 可组合：Agent 可以编程化地验证、转移、燃烧 NFT
- 标准化：CW721 是 CosmWasm 生态通用标准，钱包和浏览器天然支持
- 无需许可：Agent 自主铸造，无需中心化审批
1.2 CW721 vs CW1155
| 特性 | CW721 (ERC-721) | CW1155 (ERC-1155) |
|---|---|---|
| 唯一性 | 每个 token 唯一 | 支持半同质化 |
| 适用场景 | 成就徽章、会员卡、身份凭证 | 游戏道具、批量物品 |
| 元数据 | 每个 token 独立 | 共享元数据 + 可选独立 |
| 批量操作 | 需逐个处理 | 原生批量铸造/转账 |
| Agent 推荐 | 成就、信誉、门禁 | 大批量同类凭证 |
本指南以 CW721 为主，因为 Agent 场景中每个 NFT 代表独特成就或身份。
1.3 NFT 生命周期
铸造 (Mint) -> 持有 (Hold) -> 验证 (Verify) -> 转移 (Transfer/Burn)
| | |
v v v
Agent 铸造 用户钱包 门控检查
NFT 给用户 持有 NFT 验证持有者身份
1.4 MSG Chain 地址示例
- Agent 地址:
msg1agent2k3j4... - 用户地址:
msg1user5p6q7... - NFT 合约:
msg1nftcontract8... - 市场合约:
msg1marketplace9...
2. Agent NFT 合约
2.1 合约架构总览
Agent NFT 合约基于 cw721-base，扩展了 Agent 特定的 NFT 类型和元数据标准。
AgentNFT Contract
+-- CW721 Base (继承)
| +-- Mint / Burn / Transfer
| +-- OwnerOf / Approved
| +-- Tokens / AllTokens
+-- Agent 扩展
| +-- NftType 枚举
| +-- AchievementMetadata
| +-- MembershipMetadata
| +-- ReputationMetadata
| +-- ContributionMetadata
| +-- AccessPassMetadata
+-- 权限控制
| +-- only_minter (Agent)
| +-- only_owner
| +-- only_approved
+-- 查询扩展
+-- UserNFTs
+-- NFTsByType
+-- ReputationBadges
2.2 Cargo.toml
[package]
name = "agent-nft"
version = "0.1.0"
edition = "2021"
[lib]
crate-type = ["cdylib", "rlib"]
[dependencies]
cosmwasm-std = { version = "1.5", features = ["staking"] }
cosmwasm-storage = "1.5"
cw-storage-plus = "1.2"
cw721 = "0.18"
cw721-base = { version = "0.18", features = ["library"] }
cw-utils = "0.18"
cw2 = "0.18"
serde = { version = "1.0", features = ["derive"] }
serde-json-wasm = "0.5"
thiserror = "1.0"
url = "2.5"
uint = { version = "0.9", default-features = false }
[features]
default = ["library"]
library = []
2.3 NFT 类型枚举
use cosmwasm_std::*;
use cw721_base::*;
use serde::{Deserialize, Serialize};
/// NFT 类型 -- 决定元数据结构和铸造逻辑
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub enum NftType {
/// 成就徽章：完成特定任务后铸造
Achievement,
/// 会员资格：有时间限制或永久会员
Membership,
/// 信誉徽章：基于信誉评分铸造
ReputationBadge,
/// 贡献证明：记录对生态的贡献
ProofOfContribution,
/// 通行证：门控访问凭证
AccessPass,
/// 自定义类型，留给未来扩展
Custom(String),
}
impl NftType {
pub fn as_str(&self) -> &str {
match self {
NftType::Achievement => "achievement",
NftType::Membership => "membership",
NftType::ReputationBadge => "reputation_badge",
NftType::ProofOfContribution => "proof_of_contribution",
NftType::AccessPass => "access_pass",
NftType::Custom(s) => s,
}
}
pub fn from_str(s: &str) -> Self {
match s {
"achievement" => NftType::Achievement,
"membership" => NftType::Membership,
"reputation_badge" => NftType::ReputationBadge,
"proof_of_contribution" => NftType::ProofOfContribution,
"access_pass" => NftType::AccessPass,
_ => NftType::Custom(s.to_string()),
}
}
}
2.4 各类型元数据结构
/// 成就元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct AchievementMetadata {
/// 成就名称（人类可读）
pub title: String,
/// 成就描述
pub description: String,
/// 成就类型分类
pub achievement_type: String,
/// 完成成就时的上下文数据(JSON)
pub context: String,
/// 成就等级（1-10）
pub level: u8,
/// 达成时间戳
pub achieved_at: u64,
/// 关联的 Agent DID
pub agent_did: String,
/// 是否可升级
pub upgradeable: bool,
}
/// 会员元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct MembershipMetadata {
/// 会员等级
pub tier: String,
/// 生效时间
pub valid_from: u64,
/// 到期时间（0 表示永久）
pub valid_until: u64,
/// 会员权益列表
pub benefits: Vec<String>,
/// 是否自动续费
pub auto_renew: bool,
/// 发行方 Agent
pub issuer_agent: String,
/// 关联的支付 ID（如有）
pub payment_ref: Option<String>,
}
/// 信誉徽章元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct ReputationMetadata {
/// 信誉总分
pub total_score: u64,
/// 各维度评分
pub dimensions: Vec<ReputationDimension>,
/// 当前等级
pub level: u8,
/// 徽章在生态系统中的权重
pub weight: Decimal,
/// 最后更新时间
pub last_updated: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct ReputationDimension {
pub name: String,
pub score: u64,
pub max_score: u64,
}
/// 贡献证明元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct ContributionMetadata {
/// 贡献类型（code, data, compute, governance）
pub contribution_type: String,
/// 贡献 ID
pub contribution_id: String,
/// 贡献描述
pub description: String,
/// 贡献量
pub amount: Uint128,
/// 贡献时间范围
pub period_start: u64,
pub period_end: u64,
/// 验证者 Agent 列表
pub verifiers: Vec<String>,
}
/// 通行证元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
pub struct AccessPassMetadata {
/// 授予的权限列表
pub permissions: Vec<String>,
/// 最大使用次数（0 无限制）
pub max_uses: u32,
/// 当前使用次数
pub used_count: u32,
/// 生效时间
pub effective_at: u64,
/// 到期时间
pub expires_at: u64,
/// 是否可转让
pub transferable: bool,
}
/// 统一扩展元数据
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)]
#[serde(untagged)]
pub enum NftExtension {
Achievement(AchievementMetadata),
Membership(MembershipMetadata),
Reputation(ReputationMetadata),
Contribution(ContributionMetadata),
AccessPass(AccessPassMetadata),
}
2.5 实例化消息
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct InstantiateMsg {
/// NFT 集合名称
pub name: String,
/// NFT 集合符号
pub symbol: String,
/// 铸造者（Agent 地址）
pub minter: String,
/// Agent 去中心化身份
pub agent_did: String,
/// 集合中 NFT 的默认类型
pub nft_type: NftType,
/// 集合描述
pub description: String,
/// 集合图标 URI
pub image: Option<String>,
/// 外部链接
pub external_link: Option<String>,
/// 版税信息
pub royalty_info: Option<RoyaltyInfo>,
}
/// 版税配置
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct RoyaltyInfo {
/// 版税接收地址
pub creator: String,
/// 版税百分比（基点，如 500 = 5%）
pub royalty_basis_points: u64,
}
/// 合约实例化
pub fn instantiate(
deps: DepsMut,
_env: Env,
info: MessageInfo,
msg: InstantiateMsg,
) -> StdResult<Response> {
set_contract_version(deps.storage, "agent-nft", "0.1.0")?;
let minter = deps.api.addr_validate(&msg.minter)?;
let creator = if let Some(ref royalty) = msg.royalty_info {
Some(deps.api.addr_validate(&royalty.creator)?)
} else {
None
};
CONFIG.save(deps.storage, &Config {
name: msg.name,
symbol: msg.symbol,
minter: minter.clone(),
agent_did: msg.agent_did,
nft_type: msg.nft_type,
description: msg.description,
image: msg.image,
external_link: msg.external_link,
creator,
royalty_basis_points: msg.royalty_info
.map(|r| r.royalty_basis_points)
.unwrap_or(0),
})?;
let cw721_msg = cw721_base::InstantiateMsg {
name: msg.name,
symbol: msg.symbol,
minter: info.sender.to_string(),
};
let cw721_resp = cw721_base::instantiate(deps, _env, info, cw721_msg)?;
Ok(cw721_resp
.add_attribute("agent_did", minter.to_string())
.add_attribute("nft_type", format!("{:?}", msg.nft_type)))
}
2.6 铸造成就 NFT
/// 铸造成就 NFT
pub fn mint_achievement(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
token_id: String,
achievement_type: String,
metadata: AchievementMetadata,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.minter {
return Err(StdError::generic_err("Only the agent minter can mint NFTs"));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
if TOKENS.may_load(deps.storage, &token_id)?.is_some() {
return Err(StdError::generic_err(format!(
"Token ID {} already exists", token_id
)));
}
let extension = NftExtension::Achievement(metadata.clone());
let token = cw721_base::state::TokenInfo {
owner: recipient_addr,
approvals: vec![],
token_uri: None,
extension,
};
TOKENS.save(deps.storage, &token_id, &token)?;
NUM_TOKENS.update(deps.storage, |n| Ok(n + 1))?;
ACHIEVEMENTS.save(deps.storage, &token_id, &AchievementRecord {
token_id: token_id.clone(),
owner: recipient_addr,
achievement_type: achievement_type.clone(),
level: metadata.level,
achieved_at: metadata.achieved_at,
agent_did: config.agent_did.clone(),
})?;
USER_ACHIEVEMENTS.update(deps.storage, recipient_addr.as_bytes(), |existing| {
let mut records = existing.unwrap_or_default();
records.push(token_id.clone());
Ok::<_, StdError>(records)
})?;
Ok(Response::new()
.add_attribute("action", "mint_achievement")
.add_attribute("minter", info.sender)
.add_attribute("recipient", recipient)
.add_attribute("token_id", &token_id)
.add_attribute("achievement_type", &achievement_type)
.add_attribute("level", metadata.level.to_string())
.add_attribute("agent_did", &config.agent_did))
}
2.7 铸造会员 NFT
/// 铸造会员资格 NFT（带时间限制）
pub fn mint_membership(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
token_id: String,
metadata: MembershipMetadata,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.minter {
return Err(StdError::generic_err("Only the agent minter can mint NFTs"));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
if metadata.valid_until > 0 && metadata.valid_until <= env.block.time.seconds() {
return Err(StdError::generic_err("Membership validity must be in the future"));
}
if metadata.valid_until > 0 && metadata.valid_from >= metadata.valid_until {
return Err(StdError::generic_err("valid_from must be before valid_until"));
}
let extension = NftExtension::Membership(metadata.clone());
let token = cw721_base::state::TokenInfo {
owner: recipient_addr.clone(),
approvals: vec![],
token_uri: None,
extension,
};
TOKENS.save(deps.storage, &token_id, &token)?;
NUM_TOKENS.update(deps.storage, |n| Ok(n + 1))?;
MEMBERSHIPS.save(deps.storage, &token_id, &MembershipRecord {
token_id: token_id.clone(),
owner: recipient_addr.clone(),
tier: metadata.tier.clone(),
valid_from: metadata.valid_from,
valid_until: metadata.valid_until,
beneficiary: recipient_addr,
issuer_agent: config.agent_did.clone(),
})?;
ACTIVE_MEMBERSHIPS.update(deps.storage, recipient_addr.as_bytes(), |existing| {
let mut records = existing.unwrap_or_default();
records.push(MembershipSummary {
token_id: token_id.clone(),
tier: metadata.tier.clone(),
valid_until: metadata.valid_until,
});
Ok(records)
})?;
Ok(Response::new()
.add_attribute("action", "mint_membership")
.add_attribute("minter", info.sender)
.add_attribute("recipient", recipient)
.add_attribute("token_id", &token_id)
.add_attribute("tier", &metadata.tier)
.add_attribute("valid_until", metadata.valid_until.to_string()))
}
2.8 信誉徽章铸造
/// 铸造/升级信誉徽章
pub fn mint_reputation_badge(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
token_id: String,
metadata: ReputationMetadata,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.minter {
return Err(StdError::generic_err("Only the agent minter can mint NFTs"));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
let existing_badge = USER_REPUTATION_BADGES
.may_load(deps.storage, recipient_addr.as_bytes())?;
if let Some(existing) = existing_badge {
if existing.total_score < metadata.total_score {
TOKENS.remove(deps.storage, &existing.token_id);
NUM_TOKENS.update(deps.storage, |n| Ok(n.saturating_sub(1)))?;
let _burn_event = Event::new("burn")
.add_attribute("token_id", &existing.token_id)
.add_attribute("reason", "upgraded");
} else {
return Err(StdError::generic_err(
"User already has a badge with equal or higher score",
));
}
}
let extension = NftExtension::Reputation(metadata.clone());
let token = cw721_base::state::TokenInfo {
owner: recipient_addr.clone(),
approvals: vec![],
token_uri: None,
extension,
};
TOKENS.save(deps.storage, &token_id, &token)?;
NUM_TOKENS.update(deps.storage, |n| Ok(n + 1))?;
USER_REPUTATION_BADGES.save(deps.storage, recipient_addr.as_bytes(), &ReputationBadgeRecord {
token_id: token_id.clone(),
owner: recipient_addr.clone(),
total_score: metadata.total_score,
level: metadata.level,
last_updated: env.block.time.seconds(),
})?;
REPUTATION_RANKING.save(deps.storage, recipient_addr.as_bytes(), &metadata.total_score)?;
Ok(Response::new()
.add_attribute("action", "mint_reputation_badge")
.add_attribute("recipient", recipient)
.add_attribute("token_id", &token_id)
.add_attribute("score", metadata.total_score.to_string())
.add_attribute("level", metadata.level.to_string())
.add_attribute("is_upgrade", existing_badge.is_some().to_string()))
}
2.9 贡献证明铸造
/// 铸造贡献证明 NFT
pub fn mint_contribution_nft(
deps: DepsMut,
env: Env,
info: MessageInfo,
contributor: String,
token_id: String,
metadata: ContributionMetadata,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.minter {
return Err(StdError::generic_err("Only the agent minter can mint NFTs"));
}
let contributor_addr = deps.api.addr_validate(&contributor)?;
if metadata.verifiers.is_empty() {
return Err(StdError::generic_err("At least one verifier is required"));
}
let extension = NftExtension::Contribution(metadata.clone());
let token = cw721_base::state::TokenInfo {
owner: contributor_addr.clone(),
approvals: vec![],
token_uri: None,
extension,
};
TOKENS.save(deps.storage, &token_id, &token)?;
NUM_TOKENS.update(deps.storage, |n| Ok(n + 1))?;
CONTRIBUTIONS.save(deps.storage, &token_id, &ContributionRecord {
token_id: token_id.clone(),
contributor: contributor_addr.clone(),
contribution_type: metadata.contribution_type.clone(),
contribution_id: metadata.contribution_id.clone(),
amount: metadata.amount,
created_at: env.block.time.seconds(),
verifiers: metadata.verifiers.clone(),
})?;
USER_CONTRIBUTIONS.update(deps.storage, contributor_addr.as_bytes(), |existing| {
let mut records = existing.unwrap_or_default();
records.push(token_id.clone());
Ok::<_, StdError>(records)
})?;
Ok(Response::new()
.add_attribute("action", "mint_contribution")
.add_attribute("contributor", contributor)
.add_attribute("token_id", &token_id)
.add_attribute("contribution_type", &metadata.contribution_type)
.add_attribute("amount", metadata.amount.to_string()))
}
2.10 访问通行证铸造
/// 铸造访问通行证
pub fn mint_access_pass(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
token_id: String,
metadata: AccessPassMetadata,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
if info.sender != config.minter {
return Err(StdError::generic_err("Only the agent minter can mint NFTs"));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
if metadata.permissions.is_empty() {
return Err(StdError::generic_err("At least one permission is required"));
}
let extension = NftExtension::AccessPass(metadata.clone());
let token = cw721_base::state::TokenInfo {
owner: recipient_addr.clone(),
approvals: vec![],
token_uri: None,
extension,
};
TOKENS.save(deps.storage, &token_id, &token)?;
NUM_TOKENS.update(deps.storage, |n| Ok(n + 1))?;
ACCESS_PASSES.save(deps.storage, &token_id, &AccessPassRecord {
token_id: token_id.clone(),
owner: recipient_addr.clone(),
permissions: metadata.permissions.clone(),
expires_at: metadata.expires_at,
transferable: metadata.transferable,
})?;
for permission in &metadata.permissions {
PERMISSION_HOLDERS.update(deps.storage, permission.as_bytes(), |existing| {
let mut holders = existing.unwrap_or_default();
holders.push(recipient_addr.to_string());
Ok::<_, StdError>(holders)
})?;
}
Ok(Response::new()
.add_attribute("action", "mint_access_pass")
.add_attribute("recipient", recipient)
.add_attribute("token_id", &token_id)
.add_attribute("permissions", metadata.permissions.join(","))
.add_attribute("expires_at", metadata.expires_at.to_string()))
}
2.11 燃烧与销毁
/// 燃烧 NFT（仅铸造者或持有者可用）
pub fn burn_nft(
deps: DepsMut,
env: Env,
info: MessageInfo,
token_id: String,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
let token = TOKENS.load(deps.storage, &token_id)?;
if info.sender != config.minter && info.sender != token.owner {
return Err(StdError::generic_err(
"Only the minter or token owner can burn",
));
}
TOKENS.remove(deps.storage, &token_id);
NUM_TOKENS.update(deps.storage, |n| Ok(n.saturating_sub(1)))?;
let owner_bytes = token.owner.as_bytes();
USER_ACHIEVEMENTS.update(deps.storage, owner_bytes, |existing| {
Ok::<_, StdError>(existing
.unwrap_or_default()
.into_iter()
.filter(|id| id != &token_id)
.collect())
})?;
ACTIVE_MEMBERSHIPS.update(deps.storage, owner_bytes, |existing| {
Ok::<_, StdError>(existing
.unwrap_or_default()
.into_iter()
.filter(|m| m.token_id != token_id)
.collect())
})?;
USER_REPUTATION_BADGES.remove(deps.storage, owner_bytes);
REPUTATION_RANKING.remove(deps.storage, owner_bytes);
Ok(Response::new()
.add_attribute("action", "burn")
.add_attribute("token_id", &token_id)
.add_attribute("burner", info.sender))
}
2.12 合约状态定义
use cw_storage_plus::{Item, Map};
/// 合约级配置
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct Config {
pub name: String,
pub symbol: String,
pub minter: Addr,
pub agent_did: String,
pub nft_type: NftType,
pub description: String,
pub image: Option<String>,
pub external_link: Option<String>,
pub creator: Option<Addr>,
pub royalty_basis_points: u64,
}
pub const CONFIG: Item<Config> = Item::new("config");
pub const TOKENS: Map<&str, cw721_base::state::TokenInfo<NftExtension>> = Map::new("tokens");
pub const NUM_TOKENS: Item<u64> = Item::new("num_tokens");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct AchievementRecord {
pub token_id: String,
pub owner: Addr,
pub achievement_type: String,
pub level: u8,
pub achieved_at: u64,
pub agent_did: String,
}
pub const ACHIEVEMENTS: Map<&str, AchievementRecord> = Map::new("achievements");
pub const USER_ACHIEVEMENTS: Map<&[u8], Vec<String>> = Map::new("user_achievements");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct MembershipRecord {
pub token_id: String,
pub owner: Addr,
pub tier: String,
pub valid_from: u64,
pub valid_until: u64,
pub beneficiary: Addr,
pub issuer_agent: String,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct MembershipSummary {
pub token_id: String,
pub tier: String,
pub valid_until: u64,
}
pub const MEMBERSHIPS: Map<&str, MembershipRecord> = Map::new("memberships");
pub const ACTIVE_MEMBERSHIPS: Map<&[u8], Vec<MembershipSummary>> = Map::new("active_memberships");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct ReputationBadgeRecord {
pub token_id: String,
pub owner: Addr,
pub total_score: u64,
pub level: u8,
pub last_updated: u64,
}
pub const USER_REPUTATION_BADGES: Map<&[u8], ReputationBadgeRecord> = Map::new("user_reputation");
pub const REPUTATION_RANKING: Map<&[u8], u64> = Map::new("reputation_ranking");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct ContributionRecord {
pub token_id: String,
pub contributor: Addr,
pub contribution_type: String,
pub contribution_id: String,
pub amount: Uint128,
pub created_at: u64,
pub verifiers: Vec<String>,
}
pub const CONTRIBUTIONS: Map<&str, ContributionRecord> = Map::new("contributions");
pub const USER_CONTRIBUTIONS: Map<&[u8], Vec<String>> = Map::new("user_contributions");
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct AccessPassRecord {
pub token_id: String,
pub owner: Addr,
pub permissions: Vec<String>,
pub expires_at: u64,
pub transferable: bool,
}
pub const ACCESS_PASSES: Map<&str, AccessPassRecord> = Map::new("access_passes");
pub const PERMISSION_HOLDERS: Map<&[u8], Vec<String>> = Map::new("permission_holders");
2.13 查询函数
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum QueryMsg {
OwnerOf { token_id: String },
AllTokens { owner: String, start_after: Option<String>, limit: Option<u32> },
UserAchievements { user: String },
UserMemberships { user: String },
UserReputation { user: String },
UserContributions { user: String, contribution_type: Option<String> },
UserPermissions { user: String },
Config {},
NftType {},
ReputationRanking { limit: Option<u32> },
}
pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
match msg {
QueryMsg::OwnerOf { token_id } => to_binary(&query_owner_of(deps, token_id)?),
QueryMsg::AllTokens { owner, start_after, limit } => {
to_binary(&query_all_tokens(deps, owner, start_after, limit)?)
}
QueryMsg::UserAchievements { user } => to_binary(&query_user_achievements(deps, user)?),
QueryMsg::UserMemberships { user } => to_binary(&query_user_memberships(deps, user)?),
QueryMsg::UserReputation { user } => to_binary(&query_user_reputation(deps, user)?),
QueryMsg::UserContributions { user, contribution_type } => {
to_binary(&query_user_contributions(deps, user, contribution_type)?)
}
QueryMsg::UserPermissions { user } => to_binary(&query_user_permissions(deps, user)?),
QueryMsg::Config {} => to_binary(&CONFIG.load(deps.storage)?),
QueryMsg::NftType {} => to_binary(&CONFIG.load(deps.storage)?.nft_type),
QueryMsg::ReputationRanking { limit } => {
to_binary(&query_reputation_ranking(deps, limit.unwrap_or(50))?)
}
}
}
fn query_owner_of(deps: Deps, token_id: String) -> StdResult<OwnerOfResponse> {
let token = TOKENS.load(deps.storage, &token_id)?;
Ok(OwnerOfResponse {
owner: token.owner.to_string(),
approvals: token.approvals.iter().map(|a| Approval {
spender: a.spender.to_string(),
expires: a.expires.clone(),
}).collect(),
})
}
fn query_all_tokens(deps: Deps, owner: String, start_after: Option<String>, limit: Option<u32>) -> StdResult<Vec<String>> {
let owner_addr = deps.api.addr_validate(&owner)?;
let limit = limit.unwrap_or(30).min(100) as usize;
let tokens: Vec<String> = TOKENS
.range(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.filter(|item| item.as_ref().map(|(_, t)| t.owner == owner_addr).unwrap_or(false))
.map(|item| item.map(|(k, _)| k))
.filter_ok(|k| {
if let Some(ref start) = start_after {
k.as_str() > start.as_str()
} else {
true
}
})
.take(limit)
.collect::<StdResult<Vec<_>>>()?;
Ok(tokens)
}
fn query_user_achievements(deps: Deps, user: String) -> StdResult<Vec<AchievementRecord>> {
let user_addr = deps.api.addr_validate(&user)?;
let token_ids = USER_ACHIEVEMENTS.may_load(deps.storage, user_addr.as_bytes())?.unwrap_or_default();
token_ids.into_iter().map(|id| ACHIEVEMENTS.load(deps.storage, &id)).collect()
}
fn query_user_memberships(deps: Deps, user: String) -> StdResult<Vec<MembershipSummary>> {
let user_addr = deps.api.addr_validate(&user)?;
Ok(ACTIVE_MEMBERSHIPS.may_load(deps.storage, user_addr.as_bytes())?.unwrap_or_default())
}
fn query_user_reputation(deps: Deps, user: String) -> StdResult<Option<ReputationBadgeRecord>> {
let user_addr = deps.api.addr_validate(&user)?;
USER_REPUTATION_BADGES.may_load(deps.storage, user_addr.as_bytes())
}
fn query_user_contributions(deps: Deps, user: String, contribution_type: Option<String>) -> StdResult<Vec<ContributionRecord>> {
let user_addr = deps.api.addr_validate(&user)?;
let token_ids = USER_CONTRIBUTIONS.may_load(deps.storage, user_addr.as_bytes())?.unwrap_or_default();
let mut records: Vec<ContributionRecord> = token_ids.into_iter()
.map(|id| CONTRIBUTIONS.load(deps.storage, &id)).collect::<StdResult<Vec<_>>>()?;
if let Some(ctype) = contribution_type {
records.retain(|r| r.contribution_type == ctype);
}
Ok(records)
}
fn query_user_permissions(deps: Deps, user: String) -> StdResult<Vec<String>> {
let user_addr = deps.api.addr_validate(&user)?;
let mut permissions = vec![];
for record in ACCESS_PASSES
.range(deps.storage, None, None, cosmwasm_std::Order::Ascending)
.filter_map(|item| item.ok())
.filter(|(_, r)| r.owner == user_addr)
{
permissions.extend(record.1.permissions);
}
permissions.sort();
permissions.dedup();
Ok(permissions)
}
fn query_reputation_ranking(deps: Deps, limit: u32) -> StdResult<Vec<(String, u64)>> {
let mut rankings: Vec<(String, u64)> = REPUTATION_RANKING
.range(deps.storage, None, None, cosmwasm_std::Order::Descending)
.filter_map(|item| item.ok())
.map(|(addr_bytes, score)| (String::from_utf8_lossy(&addr_bytes).to_string(), score))
.take(limit as usize)
.collect();
rankings.sort_by(|a, b| b.1.cmp(&a.1));
Ok(rankings)
}
2.14 入口函数
#[entry_point]
pub fn execute(
deps: DepsMut,
env: Env,
info: MessageInfo,
msg: ExecuteMsg,
) -> StdResult<Response> {
match msg {
ExecuteMsg::MintAchievement { recipient, token_id, achievement_type, metadata } => {
mint_achievement(deps, env, info, recipient, token_id, achievement_type, metadata)
}
ExecuteMsg::MintMembership { recipient, token_id, metadata } => {
mint_membership(deps, env, info, recipient, token_id, metadata)
}
ExecuteMsg::MintReputationBadge { recipient, token_id, metadata } => {
mint_reputation_badge(deps, env, info, recipient, token_id, metadata)
}
ExecuteMsg::MintContributionNFT { contributor, token_id, metadata } => {
mint_contribution_nft(deps, env, info, contributor, token_id, metadata)
}
ExecuteMsg::MintAccessPass { recipient, token_id, metadata } => {
mint_access_pass(deps, env, info, recipient, token_id, metadata)
}
ExecuteMsg::Burn { token_id } => burn_nft(deps, env, info, token_id),
ExecuteMsg::TransferNft { recipient, token_id } => {
transfer_nft(deps, env, info, recipient, token_id)
}
ExecuteMsg::UpdateRoyaltyInfo { creator, royalty_basis_points } => {
update_royalty(deps, env, info, creator, royalty_basis_points)
}
}
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
#[serde(rename_all = "snake_case")]
pub enum ExecuteMsg {
MintAchievement { recipient: String, token_id: String, achievement_type: String, metadata: AchievementMetadata },
MintMembership { recipient: String, token_id: String, metadata: MembershipMetadata },
MintReputationBadge { recipient: String, token_id: String, metadata: ReputationMetadata },
MintContributionNFT { contributor: String, token_id: String, metadata: ContributionMetadata },
MintAccessPass { recipient: String, token_id: String, metadata: AccessPassMetadata },
Burn { token_id: String },
TransferNft { recipient: String, token_id: String },
UpdateRoyaltyInfo { creator: String, royalty_basis_points: u64 },
}
2.15 转移与授权
pub fn transfer_nft(
deps: DepsMut,
env: Env,
info: MessageInfo,
recipient: String,
token_id: String,
) -> StdResult<Response> {
let config = CONFIG.load(deps.storage)?;
let mut token = TOKENS.load(deps.storage, &token_id)?;
if let NftExtension::AccessPass(ref pass) = token.extension {
if !pass.transferable {
return Err(StdError::generic_err("This access pass is not transferable"));
}
}
if info.sender != token.owner {
return Err(StdError::generic_err("Only the owner can transfer"));
}
let recipient_addr = deps.api.addr_validate(&recipient)?;
let old_owner = token.owner.clone();
token.owner = recipient_addr.clone();
TOKENS.save(deps.storage, &token_id, &token)?;
USER_ACHIEVEMENTS.update(deps.storage, old_owner.as_bytes(), |existing| {
Ok::<_, StdError>(existing.unwrap_or_default().into_iter().filter(|id| id != &token_id).collect())
})?;
USER_ACHIEVEMENTS.update(deps.storage, recipient_addr.as_bytes(), |existing| {
let mut records = existing.unwrap_or_default();
records.push(token_id.clone());
Ok(records)
})?;
Ok(Response::new()
.add_attribute("action", "transfer")
.add_attribute("token_id", &token_id)
.add_attribute("from", old_owner)
.add_attribute("to", recipient))
}
3. NFT 铸造服务
3.1 Python SDK 基础
\"\"\"
Agent NFT 铸造服务
SDK for MSG Chain NFT operations
\"\"\"
import hashlib
import json
import logging
import time
import uuid
from dataclasses import dataclass, field, asdict
from datetime import datetime, timedelta
from enum import Enum
from typing import Any, Optional
import httpx
from cosmos_sdk.client.lcd import LCDClient
from cosmos_sdk.core import Coins
from cosmos_sdk.core.fee import Fee
from cosmos_sdk.core.wasm import MsgExecuteContract, MsgInstantiateContract
from cosmos_sdk.key.mnemonic import MnemonicKey
logger = logging.getLogger(__name__)
MSG_CHAIN_ID = "msg-chain-1"
MSG_LCD_URL = "https://lcd.msgchain.org"
MSG_GAS_PRICE = "1000000000000000000attoMSG"
MSG_DENOM = "umsg"
class NftType(str, Enum):
ACHIEVEMENT = "achievement"
MEMBERSHIP = "membership"
REPUTATION_BADGE = "reputation_badge"
PROOF_OF_CONTRIBUTION = "proof_of_contribution"
ACCESS_PASS = "access_pass"
@dataclass
class AgentNFTConfig:
\"\"\"Agent NFT 服务配置\"\"\"
mnemonic: str
nft_contract_addr: str = ""
marketplace_addr: str = ""
lcd_url: str = MSG_LCD_URL
chain_id: str = MSG_CHAIN_ID
gas_price: str = MSG_GAS_PRICE
gas_adjustment: float = 1.4
def get_client(self) -> LCDClient:
return LCDClient(
chain_id=self.chain_id,
url=self.lcd_url,
gas_prices=self.gas_price,
gas_adjustment=self.gas_adjustment,
)
def get_wallet(self):
key = MnemonicKey(mnemonic=self.mnemonic)
client = self.get_client()
return client.wallet(key)
3.2 元数据上传
class MetadataStorage:
\"\"\"NFT 元数据存储（IPFS / Arweave）\"\"\"
def __init__(self, storage_type: str = "ipfs", api_key: str = ""):
self.storage_type = storage_type
self.api_key = api_key
self.client = httpx.AsyncClient(timeout=30)
async def upload_metadata(self, metadata: dict) -> str:
\"\"\"上传元数据并返回 URI\"\"\"
if self.storage_type == "ipfs":
return await self._upload_to_ipfs(metadata)
elif self.storage_type == "arweave":
return await self._upload_to_arweave(metadata)
else:
return await self._upload_to_central(metadata)
async def _upload_to_ipfs(self, metadata: dict) -> str:
\"\"\"上传到 IPFS（通过 Pinata 或 Infura）\"\"\"
headers = {"Authorization": f"Bearer {self.api_key}", "Content-Type": "application/json"}
async with httpx.AsyncClient() as client:
resp = await client.post(
"https://api.pinata.cloud/pinning/pinJSONToIPFS",
json=metadata, headers=headers,
)
resp.raise_for_status()
cid = resp.json().get("IpfsHash")
return f"ipfs://{cid}"
async def _upload_to_arweave(self, metadata: dict) -> str:
\"\"\"上传到 Arweave\"\"\"
tx_data = json.dumps(metadata).encode()
tx_hash = hashlib.sha256(tx_data).hexdigest()
return f"ar://{tx_hash}"
async def _upload_to_central(self, metadata: dict) -> str:
\"\"\"上传到中心化存储（开发阶段使用）\"\"\"
async with httpx.AsyncClient() as client:
resp = await client.post("https://nft.msgchain.org/api/metadata", json=metadata)
resp.raise_for_status()
return resp.json()["uri"]
3.3 基础 Agent NFT 服务
class AgentNFTService:
\"\"\"Agent NFT 铸造服务 -- 核心逻辑\"\"\"
def __init__(self, config: AgentNFTConfig):
self.config = config
self.wallet = config.get_wallet()
self.client = config.get_client()
self.metadata_storage = MetadataStorage(storage_type="ipfs", api_key=config.mnemonic[:32])
async def _execute_contract(self, contract_addr: str, msg: dict, funds: Optional[list] = None) -> dict:
\"\"\"执行合约调用\"\"\"
execute_msg = MsgExecuteContract(
sender=self.wallet.key.acc_address,
contract=contract_addr,
msg=msg,
funds=funds or [],
)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
logger.info(f"TX sent: {result.txhash}")
return {"txhash": result.txhash, "raw_log": result.raw_log, "code": result.code}
async def _query_contract(self, contract_addr: str, msg: dict) -> Any:
\"\"\"查询合约\"\"\"
return self.client.wasm.contract_query(contract_addr, msg)
async def _generate_token_id(self, prefix: str, user_addr: str) -> str:
\"\"\"生成唯一 token_id\"\"\"
raw = f"{prefix}-{user_addr}-{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}"
return hashlib.sha256(raw.encode()).hexdigest()[:32]
3.4 铸造信誉徽章
class AgentNFTService:
async def mint_reputation_badge(
self, user_addr: str, total_score: int, dimensions: Optional[list[dict]] = None,
) -> dict:
\"\"\"
铸造信誉徽章给用户
评分等级映射:
Level 0: 0-19 基础
Level 1: 20-39 青铜
Level 2: 40-59 白银
Level 3: 60-79 黄金
Level 4: 80-99 铂金
Level 5: 100+ 钻石
\"\"\"
level = min(total_score // 20, 5)
level_names = ["基础", "青铜", "白银", "黄金", "铂金", "钻石"]
metadata = {
"name": f"信誉徽章 - {level_names[level]} Level {level}",
"description": f"在 MSG Chain 生态中获得 {total_score}/100 信誉评分",
"image": f"https://nft.msgchain.org/badges/reputation/v2/{level}.png",
"external_url": f"https://msgchain.org/badge/{user_addr}",
"attributes": [
{"trait_type": "Score", "value": total_score, "display_type": "number"},
{"trait_type": "Level", "value": level, "display_type": "number"},
{"trait_type": "Level Name", "value": level_names[level]},
{"trait_type": "Date", "value": int(time.time()), "display_type": "date"},
{"trait_type": "Max Score", "value": 100, "display_type": "number"},
],
}
if dimensions:
for dim in dimensions:
metadata["attributes"].append({
"trait_type": f"Dimension: {dim['name']}",
"value": dim["score"], "max_value": dim.get("max_score", 100),
})
token_uri = await self.metadata_storage.upload_metadata(metadata)
token_id = await self._generate_token_id("reputation", user_addr)
rep_metadata = {
"total_score": total_score, "dimensions": dimensions or [],
"level": level, "weight": str(0.1 * (level + 1)),
"last_updated": int(time.time()),
}
msg = {"mint_reputation_badge": {"recipient": user_addr, "token_id": token_id, "metadata": rep_metadata}}
result = await self._execute_contract(self.config.nft_contract_addr, msg)
return {
"token_id": token_id, "token_uri": token_uri, "level": level,
"level_name": level_names[level], "score": total_score, "txhash": result["txhash"],
}
async def get_user_reputation(self, user_addr: str) -> Optional[dict]:
msg = {"user_reputation": {"user": user_addr}}
return await self._query_contract(self.config.nft_contract_addr, msg)
async def get_reputation_ranking(self, top_n: int = 50) -> list:
msg = {"reputation_ranking": {"limit": top_n}}
return await self._query_contract(self.config.nft_contract_addr, msg)
3.5 铸造会员通行证
class AgentNFTService:
async def mint_membership_pass(
self, user_addr: str, tier: str, duration_days: int,
benefits: Optional[list[str]] = None, auto_renew: bool = False,
) -> dict:
\"\"\"
铸造时间限制会员 NFT
等级:
- "bronze": 基础会员, 30天
- "silver": 银卡会员, 90天
- "gold": 金卡会员, 180天
- "platinum": 白金会员, 365天
- "lifetime": 终身会员, 永久
\"\"\"
tier_benefits = {
"bronze": ["基础 API 调用: 1000次/天", "社区支持", "基础数据分析"],
"silver": ["API 调用: 10000次/天", "优先支持", "高级数据分析", "自定义 Webhook"],
"gold": ["API 调用: 100000次/天", "专属支持经理", "实时数据分析", "自定义集成"],
"platinum": ["无限制 API 调用", "专属支持团队", "白标解决方案", "SLA 保障"],
"lifetime": ["所有白金权益", "终身免费升级", "创始成员身份", "治理投票权"],
}
valid_from = int(time.time())
valid_until = 0 if duration_days == 0 else valid_from + (duration_days * 86400)
chosen_benefits = benefits or tier_benefits.get(tier, [])
metadata = {
"name": f"{tier.capitalize()} 会员通行证",
"description": f"MSG Chain AI Agent {tier.capitalize()} 会员资格\n时长: {'永久' if duration_days == 0 else f'{duration_days}天'}\n权益: {', '.join(chosen_benefits[:3])}...",
"image": f"https://nft.msgchain.org/membership/{tier}.png",
"attributes": [
{"trait_type": "Tier", "value": tier},
{"trait_type": "Duration (Days)", "value": duration_days},
{"trait_type": "Valid From", "value": valid_from, "display_type": "date"},
{"trait_type": "Valid Until", "value": valid_until, "display_type": "date"},
{"trait_type": "Auto Renew", "value": auto_renew},
],
}
token_uri = await self.metadata_storage.upload_metadata(metadata)
token_id = await self._generate_token_id(f"member-{tier}", user_addr)
member_metadata = {
"tier": tier, "valid_from": valid_from, "valid_until": valid_until,
"benefits": chosen_benefits, "auto_renew": auto_renew,
"issuer_agent": self.wallet.key.acc_address, "payment_ref": None,
}
msg = {"mint_membership": {"recipient": user_addr, "token_id": token_id, "metadata": member_metadata}}
result = await self._execute_contract(self.config.nft_contract_addr, msg)
return {
"token_id": token_id, "token_uri": token_uri, "tier": tier,
"valid_from": valid_from, "valid_until": valid_until,
"is_permanent": duration_days == 0, "txhash": result["txhash"],
}
async def check_membership_valid(self, user_addr: str) -> dict:
msg = {"user_memberships": {"user": user_addr}}
memberships = await self._query_contract(self.config.nft_contract_addr, msg) or []
now = int(time.time())
valid = [m for m in memberships if m["valid_until"] == 0 or m["valid_until"] > now]
expired = [m for m in memberships if m["valid_until"] != 0 and m["valid_until"] <= now]
return {
"has_valid_membership": len(valid) > 0, "active_memberships": valid,
"expired_memberships": expired,
"highest_tier": max([m["tier"] for m in valid]) if valid else None,
}
3.6 铸造成就徽章
class AgentNFTService:
async def mint_achievement_badge(
self, user_addr: str, achievement_type: str, title: str,
description: str, level: int = 1, context: Optional[dict] = None,
) -> dict:
\"\"\"
铸造成就徽章
achievement_type 预设分类:
- "first_transaction": 首次交易
- "volume_milestone": 交易量里程碑
- "accuracy_achievement": 准确率成就
- "streak_milestone": 连续签到
- "community_contributor": 社区贡献
- "referral_star": 推荐达人
- "early_adopter": 早期采用者
- "governance_participant": 治理参与
\"\"\"
type_templates = {
"first_transaction": {"name": "首次交易", "icon": "🚀"},
"volume_milestone": {"name": "交易量里程碑", "icon": "📈"},
"accuracy_achievement": {"name": "准确率成就", "icon": "🎯"},
"streak_milestone": {"name": "连续成就", "icon": "🔥"},
"community_contributor": {"name": "社区贡献者", "icon": "🤝"},
"referral_star": {"name": "推荐之星", "icon": "⭐"},
"early_adopter": {"name": "早期采用者", "icon": "🔮"},
"governance_participant": {"name": "治理参与者", "icon": "🗳️"},
}
template = type_templates.get(achievement_type, {"name": achievement_type, "icon": "🏅"})
full_title = f"{template['icon']} {title or template['name']}"
metadata = {
"name": full_title, "description": description,
"image": f"https://nft.msgchain.org/achievements/{achievement_type}/level_{level}.png",
"attributes": [
{"trait_type": "Achievement Type", "value": achievement_type},
{"trait_type": "Level", "value": level, "display_type": "number"},
{"trait_type": "Date", "value": int(time.time()), "display_type": "date"},
{"trait_type": "Icon", "value": template["icon"]},
],
}
if context:
for k, v in context.items():
metadata["attributes"].append({"trait_type": k, "value": v})
token_uri = await self.metadata_storage.upload_metadata(metadata)
token_id = await self._generate_token_id(f"ach-{achievement_type}", user_addr)
ach_metadata = {
"title": full_title, "description": description,
"achievement_type": achievement_type, "context": json.dumps(context or {}),
"level": level, "achieved_at": int(time.time()),
"agent_did": self.wallet.key.acc_address, "upgradeable": level < 5,
}
msg = {"mint_achievement": {
"recipient": user_addr, "token_id": token_id,
"achievement_type": achievement_type, "metadata": ach_metadata,
}}
result = await self._execute_contract(self.config.nft_contract_addr, msg)
return {
"token_id": token_id, "token_uri": token_uri, "title": full_title,
"achievement_type": achievement_type, "level": level, "txhash": result["txhash"],
}
async def get_user_achievements(self, user_addr: str) -> list[dict]:
msg = {"user_achievements": {"user": user_addr}}
return await self._query_contract(self.config.nft_contract_addr, msg) or []
async def verify_achievement(self, user_addr: str, achievement_type: str) -> bool:
achievements = await self.get_user_achievements(user_addr)
return any(a["achievement_type"] == achievement_type for a in achievements)
3.7 铸造贡献证明
class AgentNFTService:
async def mint_contribution_nft(
self, contribution_id: str, contributor: str, contribution_type: str,
description: str, amount: int, period_start: Optional[int] = None,
period_end: Optional[int] = None, verifiers: Optional[list[str]] = None,
) -> dict:
\"\"\"
铸造贡献证明 NFT
contribution_type:
- "code": 代码贡献（PR、提交）
- "data": 数据贡献（数据集、标注）
- "compute": 算力贡献（节点运行）
- "governance": 治理贡献（投票、提案）
- "community": 社区贡献（回答、教程）
- "security": 安全贡献（漏洞报告）
\"\"\"
type_labels = {
"code": "代码贡献", "data": "数据贡献", "compute": "算力贡献",
"governance": "治理贡献", "community": "社区贡献", "security": "安全贡献",
}
label = type_labels.get(contribution_type, contribution_type)
metadata = {
"name": f"{label} 证明",
"description": description,
"image": f"https://nft.msgchain.org/contributions/{contribution_type}.png",
"attributes": [
{"trait_type": "Contribution Type", "value": contribution_type},
{"trait_type": "Contribution ID", "value": contribution_id},
{"trait_type": "Amount", "value": amount, "display_type": "number"},
{"trait_type": "Date", "value": int(time.time()), "display_type": "date"},
{"trait_type": "Verifiers", "value": len(verifiers or [])},
],
}
token_uri = await self.metadata_storage.upload_metadata(metadata)
token_id = await self._generate_token_id(f"contrib-{contribution_type}", contributor)
contrib_metadata = {
"contribution_type": contribution_type, "contribution_id": contribution_id,
"description": description, "amount": str(amount),
"period_start": period_start or int(time.time()),
"period_end": period_end or int(time.time()),
"verifiers": verifiers or [self.wallet.key.acc_address],
}
msg = {"mint_contribution_nft": {"contributor": contributor, "token_id": token_id, "metadata": contrib_metadata}}
result = await self._execute_contract(self.config.nft_contract_addr, msg)
return {
"token_id": token_id, "token_uri": token_uri,
"contribution_type": contribution_type, "contribution_id": contribution_id,
"amount": amount, "txhash": result["txhash"],
}
async def get_user_contributions(self, user_addr: str, contribution_type: Optional[str] = None) -> list[dict]:
msg = {"user_contributions": {"user": user_addr, "contribution_type": contribution_type}}
return await self._query_contract(self.config.nft_contract_addr, msg) or []
3.8 铸造访问通行证
class AgentNFTService:
async def mint_access_pass(
self, user_addr: str, permissions: list[str], duration_days: int,
max_uses: int = 0, transferable: bool = False,
) -> dict:
\"\"\"
铸造访问通行证 NFT
permissions 可选项:
- "premium_query": 高级查询
- "unlimited_api": 无限制 API
- "data_export": 数据导出
- "batch_process": 批量处理
- "model_inference": 模型推理
- "admin_panel": 管理面板
- "audit_log": 审计日志
- "webhook_config": Webhook 配置
\"\"\"
valid_from = int(time.time())
valid_until = 0 if duration_days == 0 else valid_from + (duration_days * 86400)
metadata = {
"name": f"访问通行证 ({', '.join(permissions[:3])}{'...' if len(permissions) > 3 else ''})",
"description": f"权限: {', '.join(permissions)}\n有效期: {'永久' if duration_days == 0 else f'{duration_days}天'}\n最大使用次数: {'无限制' if max_uses == 0 else max_uses}",
"image": "https://nft.msgchain.org/access/pass.png",
"attributes": [
{"trait_type": "Permissions", "value": ", ".join(permissions)},
{"trait_type": "Duration (Days)", "value": duration_days},
{"trait_type": "Valid From", "value": valid_from, "display_type": "date"},
{"trait_type": "Valid Until", "value": valid_until, "display_type": "date"},
{"trait_type": "Max Uses", "value": max_uses, "display_type": "number"},
{"trait_type": "Transferable", "value": transferable},
],
}
token_uri = await self.metadata_storage.upload_metadata(metadata)
token_id = await self._generate_token_id("access", user_addr)
access_metadata = {
"permissions": permissions, "max_uses": max_uses, "used_count": 0,
"effective_at": valid_from, "expires_at": valid_until, "transferable": transferable,
}
msg = {"mint_access_pass": {"recipient": user_addr, "token_id": token_id, "metadata": access_metadata}}
result = await self._execute_contract(self.config.nft_contract_addr, msg)
return {
"token_id": token_id, "token_uri": token_uri, "permissions": permissions,
"valid_from": valid_from, "valid_until": valid_until,
"max_uses": max_uses, "transferable": transferable, "txhash": result["txhash"],
}
async def get_user_permissions(self, user_addr: str) -> list[str]:
msg = {"user_permissions": {"user": user_addr}}
return await self._query_contract(self.config.nft_contract_addr, msg) or []
3.9 批量铸造
class AgentNFTService:
async def batch_mint_achievements(
self, recipients: list[tuple[str, str, str, int]], achievement_type: str,
) -> list[dict]:
\"\"\"批量铸造成就\"\"\"
results = []
for user_addr, title, description, level in recipients:
try:
result = await self.mint_achievement_badge(
user_addr=user_addr, achievement_type=achievement_type,
title=title, description=description, level=level,
)
results.append(result)
logger.info(f"Minted achievement for {user_addr}: {result['token_id']}")
except Exception as e:
logger.error(f"Failed to mint for {user_addr}: {e}")
results.append({"error": str(e), "user": user_addr})
return results
async def batch_mint_contributions(self, contributions: list[dict]) -> list[dict]:
\"\"\"批量铸造贡献证明\"\"\"
results = []
for contrib in contributions:
try:
result = await self.mint_contribution_nft(
contribution_id=contrib["contribution_id"],
contributor=contrib["contributor"],
contribution_type=contrib["contribution_type"],
description=contrib["description"],
amount=contrib["amount"],
verifiers=contrib.get("verifiers"),
)
results.append(result)
except Exception as e:
logger.error(f"Failed to mint contribution: {e}")
results.append({"error": str(e), **contrib})
return results
4. NFT 市场集成
4.1 市场合约接口
\"\"\"Agent NFT 市场集成服务\"\"\"
from dataclasses import dataclass
from enum import Enum
from typing import Optional
class ListingStatus(str, Enum):
ACTIVE = "active"
SOLD = "sold"
CANCELLED = "cancelled"
AUCTION = "auction"
@dataclass
class Listing:
token_id: str
contract_addr: str
seller: str
price: int
denom: str = MSG_DENOM
status: ListingStatus = ListingStatus.ACTIVE
created_at: int = 0
def price_display(self) -> str:
return f"{self.price / 1_000_000:,.4f} MSG"
@dataclass
class Auction:
token_id: str
contract_addr: str
seller: str
reserve_price: int
highest_bid: int = 0
highest_bidder: str = ""
start_time: int = 0
end_time: int = 0
status: str = "active"
4.2 市场服务
class NFTMarketplace:
\"\"\"NFT 市场集成服务\"\"\"
def __init__(self, config: AgentNFTConfig, marketplace_addr: str):
self.config = config
self.wallet = config.get_wallet()
self.client = config.get_client()
self.marketplace_addr = marketplace_addr
self.nft_addr = config.nft_contract_addr
async def _execute(self, msg: dict, funds: Optional[list] = None) -> dict:
execute_msg = MsgExecuteContract(
sender=self.wallet.key.acc_address,
contract=self.marketplace_addr,
msg=msg, funds=funds or [],
)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {"txhash": result.txhash, "raw_log": result.raw_log}
async def approve_marketplace(self, token_id: str) -> dict:
msg = {"approve": {"spender": self.marketplace_addr, "token_id": token_id, "expires": {"never": {}}}}
return await self._execute_cw721(msg)
async def _execute_cw721(self, msg: dict) -> dict:
execute_msg = MsgExecuteContract(
sender=self.wallet.key.acc_address, contract=self.nft_addr, msg=msg,
)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {"txhash": result.txhash}
4.3 挂单与取消
class NFTMarketplace:
async def list_nft(self, token_id: str, price: int, denom: str = MSG_DENOM) -> dict:
\"\"\"
挂单出售 NFT
price 以 umsg 为单位（1 MSG = 1,000,000 umsg）
例如: 50 MSG = 50_000_000 umsg
\"\"\"
await self.approve_marketplace(token_id)
msg = {"list_nft": {"token_id": token_id, "contract_addr": self.nft_addr, "price": str(price), "denom": denom}}
result = await self._execute(msg)
return {
"action": "list", "token_id": token_id, "price": price,
"price_display": f"{price / 1_000_000:,.4f} MSG",
"denom": denom, "seller": self.wallet.key.acc_address,
"marketplace": self.marketplace_addr, "txhash": result["txhash"],
}
async def cancel_listing(self, token_id: str) -> dict:
msg = {"cancel_listing": {"token_id": token_id, "contract_addr": self.nft_addr}}
result = await self._execute(msg)
return {"action": "cancel_listing", "token_id": token_id, "txhash": result["txhash"]}
async def update_price(self, token_id: str, new_price: int) -> dict:
msg = {"update_price": {"token_id": token_id, "contract_addr": self.nft_addr, "new_price": str(new_price)}}
result = await self._execute(msg)
return {
"action": "update_price", "token_id": token_id, "new_price": new_price,
"new_price_display": f"{new_price / 1_000_000:,.4f} MSG", "txhash": result["txhash"],
}
4.4 购买 NFT
class NFTMarketplace:
async def buy_nft(self, token_id: str, price: int, denom: str = MSG_DENOM) -> dict:
funds = [{"denom": denom, "amount": str(price)}]
msg = {"buy_nft": {"token_id": token_id, "contract_addr": self.nft_addr}}
result = await self._execute(msg, funds=funds)
return {
"action": "buy", "token_id": token_id, "price": price,
"price_display": f"{price / 1_000_000:,.4f} MSG",
"buyer": self.wallet.key.acc_address, "txhash": result["txhash"],
}
async def buy_nft_for(self, token_id: str, price: int, recipient: str, denom: str = MSG_DENOM) -> dict:
funds = [{"denom": denom, "amount": str(price)}]
msg = {"buy_nft_for": {"token_id": token_id, "contract_addr": self.nft_addr, "recipient": recipient}}
result = await self._execute(msg, funds=funds)
return {"action": "buy_for", "token_id": token_id, "recipient": recipient, "price": price, "txhash": result["txhash"]}
async def batch_buy(self, purchases: list[tuple[str, str, int]]) -> list[dict]:
results = []
for token_id, contract_addr, price in purchases:
try:
result = await self.buy_nft(token_id, price)
results.append(result)
except Exception as e:
logger.error(f"Failed to buy {token_id}: {e}")
results.append({"token_id": token_id, "error": str(e)})
return results
4.5 竞拍
class NFTMarketplace:
async def auction_nft(
self, token_id: str, reserve_price: int, duration_seconds: int,
denom: str = MSG_DENOM, start_price: Optional[int] = None,
) -> dict:
await self.approve_marketplace(token_id)
msg = {
"start_auction": {
"token_id": token_id, "contract_addr": self.nft_addr,
"reserve_price": str(reserve_price),
"start_price": str(start_price or reserve_price),
"duration": duration_seconds, "denom": denom,
}
}
result = await self._execute(msg)
return {
"action": "start_auction", "token_id": token_id,
"reserve_price": reserve_price, "start_price": start_price or reserve_price,
"duration": duration_seconds, "end_time": int(time.time()) + duration_seconds,
"txhash": result["txhash"],
}
async def place_bid(self, token_id: str, bid_amount: int, denom: str = MSG_DENOM) -> dict:
funds = [{"denom": denom, "amount": str(bid_amount)}]
msg = {"place_bid": {"token_id": token_id, "contract_addr": self.nft_addr}}
result = await self._execute(msg, funds=funds)
return {
"action": "place_bid", "token_id": token_id,
"bid_amount": bid_amount, "bidder": self.wallet.key.acc_address,
"txhash": result["txhash"],
}
async def cancel_auction(self, token_id: str) -> dict:
msg = {"cancel_auction": {"token_id": token_id, "contract_addr": self.nft_addr}}
result = await self._execute(msg)
return {"action": "cancel_auction", "token_id": token_id, "txhash": result["txhash"]}
async def claim_auction(self, token_id: str) -> dict:
msg = {"claim_auction": {"token_id": token_id, "contract_addr": self.nft_addr, "claimant": self.wallet.key.acc_address}}
result = await self._execute(msg)
return {"action": "claim_auction", "token_id": token_id, "claimant": self.wallet.key.acc_address, "txhash": result["txhash"]}
4.6 查询市场数据
class NFTMarketplace:
async def get_listings(self, contract_addr: Optional[str] = None, seller: Optional[str] = None, status: Optional[ListingStatus] = None) -> list[Listing]:
msg: dict = {"listings": {}}
if contract_addr: msg["listings"]["contract_addr"] = contract_addr
if seller: msg["listings"]["seller"] = seller
if status: msg["listings"]["status"] = status.value
raw = self.client.wasm.contract_query(self.marketplace_addr, msg)
return [Listing(**l) for l in raw]
async def get_auctions(self, status: Optional[str] = None) -> list[Auction]:
msg = {"auctions": {}}
if status: msg["auctions"]["status"] = status
raw = self.client.wasm.contract_query(self.marketplace_addr, msg)
return [Auction(**a) for a in raw]
async def get_nft_price_history(self, token_id: str) -> list[dict]:
msg = {"price_history": {"token_id": token_id}}
return self.client.wasm.contract_query(self.marketplace_addr, msg) or []
async def get_market_stats(self) -> dict:
stats = self.client.wasm.contract_query(self.marketplace_addr, {"market_stats": {}}) or {}
if stats:
stats["total_volume_display"] = f"{int(stats.get('total_volume', 0)) / 1_000_000:,.2f} MSG"
return stats
4.7 市场事件监控
class NFTMarketplaceMonitor:
def __init__(self, lcd_url: str, marketplace_addr: str, poll_interval: int = 10):
self.client = LCDClient(chain_id=MSG_CHAIN_ID, url=lcd_url)
self.marketplace_addr = marketplace_addr
self.poll_interval = poll_interval
self.last_height = 0
async def poll_events(self, callback):
while True:
try:
height = self.client.block_height()
if self.last_height == 0:
self.last_height = height
continue
events = self.client.search_txs(
f"wasm._contract_address='{self.marketplace_addr}'",
min_height=self.last_height, max_height=height,
)
for tx in events:
for event in tx.logs[0].events:
if event.type == "wasm":
parsed = self._parse_event(event)
if parsed:
await callback(parsed)
self.last_height = height
except Exception as e:
logger.error(f"Poll error: {e}")
await asyncio.sleep(self.poll_interval)
def _parse_event(self, event) -> Optional[dict]:
attrs = {a.key: a.value for a in event.attributes}
action = attrs.get("action")
if action in ("list_nft", "buy_nft", "cancel_listing", "start_auction", "place_bid", "claim_auction"):
return {
"action": action, "token_id": attrs.get("token_id"), "seller": attrs.get("seller"),
"buyer": attrs.get("buyer"), "price": attrs.get("price"), "bidder": attrs.get("bidder"),
"block_height": attrs.get("_tx.height"),
}
return None
5. NFT 门控访问
5.1 门控服务核心
\"\"\"NFT 门控访问服务\"\"\"
from enum import Enum
from typing import Optional
class FeatureAccess(str, Enum):
PUBLIC = "public"
NFT_GATED = "nft_gated"
TIERED = "tiered"
TIME_LOCKED = "time_locked"
@dataclass
class GatingRule:
feature: str
access_type: FeatureAccess
collection: str
min_quantity: int = 1
required_tier: Optional[str] = None
required_permission: Optional[str] = None
max_age_days: Optional[int] = None
class NFTGatedAccess:
def __init__(self, config: AgentNFTConfig):
self.config = config
self.client = config.get_client()
self.nft_addr = config.nft_contract_addr
self.GATING_RULES: dict[str, GatingRule] = {
"premium_query": GatingRule(
feature="premium_query", access_type=FeatureAccess.NFT_GATED,
collection=config.nft_contract_addr, min_quantity=1, required_permission="premium_query",
),
"unlimited_api": GatingRule(
feature="unlimited_api", access_type=FeatureAccess.NFT_GATED,
collection=config.nft_contract_addr, min_quantity=1, required_permission="unlimited_api",
),
"data_export": GatingRule(
feature="data_export", access_type=FeatureAccess.TIERED,
collection=config.nft_contract_addr, min_quantity=1,
required_tier="silver", required_permission="data_export",
),
"batch_process": GatingRule(
feature="batch_process", access_type=FeatureAccess.NFT_GATED,
collection=config.nft_contract_addr, min_quantity=1, required_permission="batch_process",
),
"model_inference": GatingRule(
feature="model_inference", access_type=FeatureAccess.TIERED,
collection=config.nft_contract_addr, min_quantity=1,
required_tier="gold", required_permission="model_inference",
),
"admin_panel": GatingRule(
feature="admin_panel", access_type=FeatureAccess.TIERED,
collection=config.nft_contract_addr, min_quantity=1,
required_tier="platinum", required_permission="admin_panel",
),
"webhook_config": GatingRule(
feature="webhook_config", access_type=FeatureAccess.TIERED,
collection=config.nft_contract_addr, min_quantity=1,
required_tier="silver", required_permission="webhook_config",
),
}
5.2 访问检查
class NFTGatedAccess:
async def check_access(self, user: str, feature: str) -> bool:
rule = self.GATING_RULES.get(feature)
if not rule:
return True
if rule.access_type == FeatureAccess.PUBLIC:
return True
user_tokens = await self._query_user_tokens(user)
if not user_tokens:
return False
if rule.access_type == FeatureAccess.NFT_GATED:
return await self._check_nft_gated(user, user_tokens, rule)
elif rule.access_type == FeatureAccess.TIERED:
return await self._check_tiered(user, user_tokens, rule)
elif rule.access_type == FeatureAccess.TIME_LOCKED:
return await self._check_time_locked(user, user_tokens, rule)
return False
async def _query_user_tokens(self, user: str) -> list[str]:
try:
return self.client.wasm.contract_query(self.nft_addr, {"all_tokens": {"owner": user, "limit": 100}}) or []
except Exception:
return []
async def _check_nft_gated(self, user: str, tokens: list[str], rule: GatingRule) -> bool:
if len(tokens) < rule.min_quantity:
return False
if rule.required_permission:
count = 0
for token_id in tokens:
if await self._check_token_permission(token_id, rule.required_permission):
count += 1
if count >= rule.min_quantity:
return True
return count >= rule.min_quantity
return len(tokens) >= rule.min_quantity
async def _check_tiered(self, user: str, tokens: list[str], rule: GatingRule) -> bool:
tier_order = {"bronze": 0, "silver": 1, "gold": 2, "platinum": 3, "lifetime": 4}
required_level = tier_order.get(rule.required_tier, 0)
memberships = self.client.wasm.contract_query(self.nft_addr, {"user_memberships": {"user": user}}) or []
now = int(time.time())
for m in memberships:
if m["valid_until"] != 0 and m["valid_until"] < now:
continue
user_level = tier_order.get(m["tier"], -1)
if user_level >= required_level:
if rule.required_permission:
return await self._check_token_permission(m["token_id"], rule.required_permission)
return True
return False
async def _check_time_locked(self, user: str, tokens: list[str], rule: GatingRule) -> bool:
now = int(time.time())
memberships = self.client.wasm.contract_query(self.nft_addr, {"user_memberships": {"user": user}}) or []
for m in memberships:
if m["valid_until"] == 0 or m["valid_until"] > now:
return True
return False
async def _check_token_permission(self, token_id: str, required_permission: str) -> bool:
try:
token_info = self.client.wasm.contract_query(self.nft_addr, {"owner_of": {"token_id": token_id}})
if not token_info:
return False
owner = token_info.get("owner", "")
perms = self.client.wasm.contract_query(self.nft_addr, {"user_permissions": {"user": owner}}) or []
return required_permission in perms
except Exception:
return False
5.3 高速缓存访问控制
class CachedNFTGatedAccess(NFTGatedAccess):
def __init__(self, config: AgentNFTConfig, redis_url: str = "redis://localhost:6379"):
super().__init__(config)
import aioredis
self.redis = aioredis.from_url(redis_url)
self.cache_ttl = 300
async def check_access(self, user: str, feature: str) -> bool:
cache_key = f"access:{user}:{feature}"
cached = await self.redis.get(cache_key)
if cached is not None:
return cached == b"true"
result = await super().check_access(user, feature)
await self.redis.setex(cache_key, self.cache_ttl, str(result).lower())
return result
async def invalidate_user_cache(self, user: str):
pattern = f"access:{user}:*"
keys = await self.redis.keys(pattern)
if keys:
await self.redis.delete(*keys)
async def batch_check_access(self, user: str, features: list[str]) -> dict[str, bool]:
return {f: await self.check_access(user, f) for f in features}
5.4 Agent 端门控中间件
class GatingMiddleware:
def __init__(self, gated_access: NFTGatedAccess):
self.gated_access = gated_access
def require_nft(self, feature: str):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
user = kwargs.get("user") or args[0].get("user")
if not user:
return {"error": "User address required"}, 401
has_access = await self.gated_access.check_access(user, feature)
if not has_access:
return {
"error": "Access denied", "feature": feature,
"required_nft": f"Hold a valid {feature} pass",
"upgrade_url": f"https://msgchain.org/mint/{feature}",
}, 403
return await func(*args, **kwargs)
return wrapper
return decorator
async def get_access_status(self, user: str) -> dict:
features = list(self.gated_access.GATING_RULES.keys())
access_map = await self.gated_access.batch_check_access(user, features)
return {
"user": user, "access": access_map,
"granted": [f for f, a in access_map.items() if a],
"denied": [f for f, a in access_map.items() if not a],
}
5.5 时间限制处理
class NFTGatedAccess:
async def check_access_with_usage(self, user: str, feature: str) -> dict:
rule = self.GATING_RULES.get(feature)
if not rule:
return {"allowed": True, "reason": "public_feature"}
passes = self.client.wasm.contract_query(self.nft_addr, {"user_permissions": {"user": user}}) or []
if not passes:
return {"allowed": False, "reason": "no_access_pass"}
now = int(time.time())
best_pass = None
for access_pass_id in passes:
access_info = self.client.wasm.contract_query(self.nft_addr, {"access_pass": {"token_id": access_pass_id}})
if not access_info:
continue
if access_info["expires_at"] != 0 and access_info["expires_at"] < now:
continue
if access_info["max_uses"] > 0 and access_info["used_count"] >= access_info["max_uses"]:
continue
if rule.required_permission and rule.required_permission not in access_info.get("permissions", []):
continue
best_pass = access_info
break
if not best_pass:
return {"allowed": False, "reason": "no_valid_pass"}
remaining_uses = best_pass["max_uses"] - best_pass["used_count"] if best_pass["max_uses"] > 0 else -1
remaining_days = (best_pass["expires_at"] - now) // 86400 if best_pass["expires_at"] > 0 else -1
return {
"allowed": True, "token_id": best_pass["token_id"],
"remaining_uses": remaining_uses, "remaining_days": remaining_days,
"permissions": best_pass.get("permissions", []), "expires_at": best_pass.get("expires_at", 0),
}
6. NFT 与信誉系统集成
6.1 信誉积分引擎
\"\"\"NFT 信誉系统集成\"\"\"
from dataclasses import dataclass, field
@dataclass
class ReputationConfig:
categories: dict = field(default_factory=lambda: {
"trading": {"weight": 0.3, "max_score": 100},
"contribution": {"weight": 0.25, "max_score": 100},
"governance": {"weight": 0.2, "max_score": 100},
"community": {"weight": 0.15, "max_score": 100},
"longevity": {"weight": 0.1, "max_score": 100},
})
badge_levels: list = field(default_factory=lambda: [
{"level": 0, "min_score": 0, "name": "基础成员"},
{"level": 1, "min_score": 20, "name": "青铜成员"},
{"level": 2, "min_score": 40, "name": "白银成员"},
{"level": 3, "min_score": 60, "name": "黄金成员"},
{"level": 4, "min_score": 80, "name": "铂金成员"},
{"level": 5, "min_score": 100, "name": "钻石成员"},
])
nft_contract: str = ""
agent_addr: str = ""
class ReputationNFT:
def __init__(self, config: AgentNFTConfig, rep_config: Optional[ReputationConfig] = None):
self.config = config
self.wallet = config.get_wallet()
self.client = config.get_client()
self.nft_service = AgentNFTService(config)
self.rep_config = rep_config or ReputationConfig()
def _calculate_total_score(self, category_scores: dict[str, int]) -> int:
total = 0
for category, score in category_scores.items():
cat_config = self.rep_config.categories.get(category, {"weight": 0, "max_score": 100})
total += score * cat_config["weight"]
return min(int(total), 100)
def _get_level(self, total_score: int) -> dict:
for badge in reversed(self.rep_config.badge_levels):
if total_score >= badge["min_score"]:
return badge
return self.rep_config.badge_levels[0]
6.2 发放信誉 NFT
class ReputationNFT:
async def issue_reputation_nft(self, user: str, categories_scores: dict[str, int]) -> dict:
total_score = self._calculate_total_score(categories_scores)
badge_level = self._get_level(total_score)
existing = await self.nft_service.get_user_reputation(user)
if existing:
existing_score = existing.get("total_score", 0)
if existing_score >= total_score:
logger.info(f"User {user} already has badge with score {existing_score} >= {total_score}, skipping")
return {"action": "skipped", "reason": "existing_badge_equal_or_higher", "existing_score": existing_score, "new_score": total_score}
logger.info(f"Upgrading badge for {user}: {existing_score} -> {total_score}")
await self._execute_on_nft({"burn": {"token_id": existing["token_id"]}})
dimensions = [
{"name": cat, "score": score, "max_score": self.rep_config.categories.get(cat, {}).get("max_score", 100)}
for cat, score in categories_scores.items()
]
result = await self.nft_service.mint_reputation_badge(user_addr=user, total_score=total_score, dimensions=dimensions)
await self._update_on_chain_reputation(user, categories_scores, total_score)
return {
"action": "issued" if not existing else "upgraded", "user": user,
"previous_score": existing.get("total_score") if existing else None,
"new_score": total_score, "level": badge_level,
"token_id": result["token_id"], "txhash": result["txhash"],
}
async def _execute_on_nft(self, msg: dict) -> dict:
execute_msg = MsgExecuteContract(sender=self.wallet.key.acc_address, contract=self.config.nft_contract_addr, msg=msg)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {"txhash": result.txhash}
async def _update_on_chain_reputation(self, user: str, category_scores: dict[str, int], total_score: int) -> None:
rep_update_msg = {
"update_reputation": {"user": user, "scores": category_scores, "total_score": total_score, "updated_at": int(time.time())},
}
try:
await self._execute_on_nft(rep_update_msg)
logger.info(f"On-chain reputation updated for {user}")
except Exception as e:
logger.error(f"Failed to update on-chain reputation: {e}")
6.3 成就验证
class ReputationNFT:
async def verify_achievement(self, user: str, achievement_type: str) -> bool:
achievements = await self.nft_service.get_user_achievements(user)
return any(a["achievement_type"] == achievement_type for a in achievements)
async def verify_multiple_achievements(self, user: str, required_achievements: list[str], require_all: bool = True) -> dict:
achievements = await self.nft_service.get_user_achievements(user)
held_types = {a["achievement_type"] for a in achievements}
results = {}
missing = []
for ach in required_achievements:
has = ach in held_types
results[ach] = has
if not has:
missing.append(ach)
verified = len(missing) == 0 if require_all else len(required_achievements) - len(missing) > 0
return {"verified": verified, "results": results, "missing": missing, "held_count": len(held_types), "required_count": len(required_achievements)}
async def get_achievement_summary(self, user: str) -> dict:
achievements = await self.nft_service.get_user_achievements(user)
by_type = {}
for ach in achievements:
atype = ach["achievement_type"]
by_type.setdefault(atype, []).append(ach)
return {
"user": user, "total_achievements": len(achievements),
"by_type": {k: len(v) for k, v in by_type.items()},
"highest_level": max((a.get("level", 0) for a in achievements), default=0),
"achievements": achievements,
}
6.4 信誉数据查询
class ReputationNFT:
async def get_user_reputation_profile(self, user: str) -> dict:
badge = await self.nft_service.get_user_reputation(user)
achievements = await self.nft_service.get_user_achievements(user)
contributions = await self.nft_service.get_user_contributions(user)
memberships = await self.nft_service.check_membership_valid(user)
permissions = await self.nft_service.get_user_permissions(user)
rep_score = badge["total_score"] if badge else 0
ach_score = min(len(achievements) * 5, 30)
contrib_score = min(sum(int(c.get("amount", 0)) for c in contributions) // 1000, 20)
composite = min(rep_score + ach_score + contrib_score, 100)
level = self._get_level(composite)
return {
"user": user, "composite_score": composite, "level": level,
"components": {
"reputation_badge": {"score": rep_score, "exists": badge is not None},
"achievements": {"score": ach_score, "count": len(achievements)},
"contributions": {"score": contrib_score, "count": len(contributions)},
},
"badge": badge, "achievements": achievements, "contributions": contributions,
"memberships": memberships, "permissions": permissions,
}
async def get_leaderboard(self, category: Optional[str] = None, top_n: int = 20) -> list[dict]:
msg = {"reputation_ranking": {"limit": top_n}}
ranking = await self.nft_service._query_contract(self.config.nft_contract_addr, msg)
enriched = []
for user_addr, score in ranking[:top_n]:
badge = await self.nft_service.get_user_reputation(user_addr)
level = self._get_level(score)
enriched.append({
"user": user_addr, "score": score,
"level": level["name"], "level_number": level["level"],
"badge_token_id": badge["token_id"] if badge else None,
})
return enriched
6.5 信誉数据激励
class ReputationNFT:
async def check_and_reward_milestone(self, user: str, event_type: str, event_data: dict) -> Optional[dict]:
milestones = {
"transaction": [
{"name": "first_transaction", "threshold": 1, "title": "首次交易", "level": 1},
{"name": "volume_milestone", "threshold": 100, "title": "百笔交易", "level": 2},
{"name": "volume_milestone", "threshold": 1000, "title": "千笔交易", "level": 3},
{"name": "volume_milestone", "threshold": 10000, "title": "万笔交易", "level": 4},
],
"contribution": [
{"name": "first_contribution", "threshold": 1, "title": "首次贡献", "level": 1},
{"name": "community_contributor", "threshold": 10, "title": "十次贡献", "level": 2},
{"name": "community_contributor", "threshold": 50, "title": "贡献达人", "level": 3},
],
"governance_vote": [
{"name": "governance_participant", "threshold": 1, "title": "首次投票", "level": 1},
{"name": "governance_participant", "threshold": 10, "title": "积极选民", "level": 2},
],
"referral": [
{"name": "referral_star", "threshold": 1, "title": "推荐新秀", "level": 1},
{"name": "referral_star", "threshold": 5, "title": "推荐达人", "level": 2},
{"name": "referral_star", "threshold": 20, "title": "推荐之星", "level": 3},
],
"streak": [
{"name": "streak_milestone", "threshold": 7, "title": "连续7天", "level": 1},
{"name": "streak_milestone", "threshold": 30, "title": "连续30天", "level": 2},
{"name": "streak_milestone", "threshold": 100, "title": "连续100天", "level": 3},
],
}
event_milestones = milestones.get(event_type, [])
if not event_milestones:
return None
current_count = event_data.get(f"count_{event_type}", 0)
earned = []
for ms in event_milestones:
if current_count >= ms["threshold"]:
has = await self.verify_achievement(user, ms["name"])
if not has:
achievement = await self.nft_service.mint_achievement_badge(
user_addr=user, achievement_type=ms["name"], title=ms["title"],
description=f"达成 {ms['threshold']} {event_type} 里程碑",
level=ms["level"],
context={"event_type": event_type, "threshold": ms["threshold"], "current_count": current_count},
)
earned.append(achievement)
cat = {
"transaction": "trading", "contribution": "contribution",
"governance_vote": "governance", "referral": "community", "streak": "longevity",
}.get(event_type, "community")
profile = await self.get_user_reputation_profile(user)
current_scores = {}
for c in self.rep_config.categories:
current_scores[c] = 0
if profile.get("badge"):
for dim in profile["badge"].get("dimensions", []):
current_scores[dim["name"]] = dim["score"]
current_scores[cat] = min(current_scores.get(cat, 0) + 5, self.rep_config.categories[cat]["max_score"])
await self.issue_reputation_nft(user, current_scores)
return {"earned": earned} if earned else None
7. NFT 收益与版税
7.1 版税合约逻辑
/// 版税与收益分配模块
use cosmwasm_std::{
coin, ensure, Addr, BankMsg, Coin, Decimal256, DepsMut,
Env, MessageInfo, Response, StdError, StdResult, Uint128,
};
use cw_storage_plus::{Item, Map};
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct RoyaltyConfig {
pub creator: Addr,
pub royalty_basis_points: u64,
pub total_royalty_collected: Uint128,
pub collection_count: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct SaleInfo {
pub token_id: String,
pub contract_addr: Addr,
pub seller: Addr,
pub buyer: Addr,
pub price: Uint128,
pub denom: String,
pub sold_at: u64,
}
#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, JsonSchema)]
pub struct RoyaltyEarning {
pub recipient: Addr,
pub token_id: String,
pub amount: Uint128,
pub denom: String,
pub sale_txhash: String,
pub earned_at: u64,
}
pub const ROYALTY_CONFIG: Item<RoyaltyConfig> = Item::new("royalty_config");
pub const ROYALTY_EARNINGS: Map<&[u8], Vec<RoyaltyEarning>> = Map::new("royalty_earnings");
pub const SALE_HISTORY: Map<&str, SaleInfo> = Map::new("sale_history");
pub fn calculate_royalty(price: Uint128, royalty_basis_points: u64) -> StdResult<Uint128> {
if royalty_basis_points > 10000 {
return Err(StdError::generic_err("Royalty basis points cannot exceed 10000"));
}
Ok(price.multiply_ratio(royalty_basis_points, 10000u64))
}
pub fn handle_sale(deps: DepsMut, env: Env, info: MessageInfo, sale: SaleInfo) -> StdResult<Response> {
let config = ROYALTY_CONFIG.load(deps.storage)?;
let royalty_amount = calculate_royalty(sale.price, config.royalty_basis_points)?;
let seller_net = sale.price.checked_sub(royalty_amount)
.map_err(|_| StdError::generic_err("Price overflow"))?;
let royalty_msg = BankMsg::Send {
to_address: config.creator.to_string(),
amount: vec![Coin { denom: sale.denom.clone(), amount: royalty_amount }],
};
let seller_msg = BankMsg::Send {
to_address: sale.seller.to_string(),
amount: vec![Coin { denom: sale.denom.clone(), amount: seller_net }],
};
let mut updated_config = config.clone();
updated_config.total_royalty_collected = config.total_royalty_collected.checked_add(royalty_amount)?;
updated_config.collection_count += 1;
ROYALTY_CONFIG.save(deps.storage, &updated_config)?;
let earning = RoyaltyEarning {
recipient: config.creator.clone(), token_id: sale.token_id.clone(),
amount: royalty_amount, denom: sale.denom.clone(),
sale_txhash: env.transaction.as_ref().map(|t| t.hash.clone()).unwrap_or_default(),
earned_at: env.block.time.seconds(),
};
let mut earnings = ROYALTY_EARNINGS.may_load(deps.storage, config.creator.as_bytes())?.unwrap_or_default();
earnings.push(earning);
ROYALTY_EARNINGS.save(deps.storage, config.creator.as_bytes(), &earnings)?;
SALE_HISTORY.save(deps.storage, &sale.token_id, &sale)?;
Ok(Response::new()
.add_message(royalty_msg).add_message(seller_msg)
.add_attribute("action", "handle_sale")
.add_attribute("token_id", &sale.token_id)
.add_attribute("seller", &sale.seller)
.add_attribute("buyer", &sale.buyer)
.add_attribute("price", sale.price.to_string())
.add_attribute("royalty_amount", royalty_amount.to_string())
.add_attribute("seller_net", seller_net.to_string())
.add_attribute("royalty_percentage", config.royalty_basis_points.to_string()))
}
pub fn update_royalty(deps: DepsMut, env: Env, info: MessageInfo, creator: String, royalty_basis_points: u64) -> StdResult<Response> {
let config = ROYALTY_CONFIG.load(deps.storage)?;
let contract_config = CONFIG.load(deps.storage)?;
if info.sender != contract_config.minter && info.sender != config.creator {
return Err(StdError::generic_err("Only the minter or creator can update royalty"));
}
if royalty_basis_points > 10000 {
return Err(StdError::generic_err("Royalty cannot exceed 10000 bps"));
}
ROYALTY_CONFIG.save(deps.storage, &RoyaltyConfig {
creator: deps.api.addr_validate(&creator)?,
royalty_basis_points,
total_royalty_collected: config.total_royalty_collected,
collection_count: config.collection_count,
})?;
Ok(Response::new()
.add_attribute("action", "update_royalty")
.add_attribute("creator", creator)
.add_attribute("royalty_basis_points", royalty_basis_points.to_string()))
}
7.2 Python 版税管理服务
\"\"\"版税与收益管理服务\"\"\"
from decimal import Decimal
from typing import Optional
class RoyaltyManager:
def __init__(self, config: AgentNFTConfig):
self.config = config
self.wallet = config.get_wallet()
self.client = config.get_client()
async def set_royalty(self, creator: str, royalty_basis_points: int) -> dict:
msg = {"update_royalty_info": {"creator": creator, "royalty_basis_points": royalty_basis_points}}
execute_msg = MsgExecuteContract(sender=self.wallet.key.acc_address, contract=self.config.nft_contract_addr, msg=msg)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {
"action": "set_royalty", "creator": creator, "royalty_basis_points": royalty_basis_points,
"royalty_percentage": f"{royalty_basis_points / 100}%", "txhash": result.txhash,
}
async def get_royalty_info(self) -> dict:
info = self.client.wasm.contract_query(self.config.nft_contract_addr, {"config": {}})
return {
"creator": info.get("creator"), "royalty_basis_points": info.get("royalty_basis_points", 0),
"royalty_percentage": f"{info.get('royalty_basis_points', 0) / 100}%",
}
async def get_royalty_earnings(self, address: Optional[str] = None) -> list[dict]:
target = address or self.wallet.key.acc_address
return self.client.wasm.contract_query(self.config.nft_contract_addr, {"royalty_earnings": {"address": target, "limit": 100}}) or []
async def get_total_royalty_collected(self) -> dict:
config = self.client.wasm.contract_query(self.config.nft_contract_addr, {"config": {}})
total = int(config.get("total_royalty_collected", 0))
return {"total_collected": str(total), "total_collected_display": f"{total / 1_000_000:,.2f} MSG", "sale_count": config.get("collection_count", 0)}
7.3 二次销售收益分配
class RoyaltyManager:
async def distribute_secondary_sale(self, token_id: str, sale_price: int, buyer: str, seller: str) -> dict:
info = await self.get_royalty_info()
basis_points = info.get("royalty_basis_points", 0)
royalty_amount = sale_price * basis_points // 10000
market_fee = sale_price * 200 // 10000
seller_amount = sale_price - royalty_amount - market_fee
record = {
"token_id": token_id, "price": sale_price, "royalty": royalty_amount,
"market_fee": market_fee, "seller_net": seller_amount,
"buyer": buyer, "seller": seller, "timestamp": int(time.time()),
}
logger.info(f"Secondary sale: token={token_id} price={sale_price} royalty={royalty_amount} seller={seller_amount}")
return record
async def get_sale_history(self, token_id: Optional[str] = None) -> list[dict]:
msg = {"sale_history": {"token_id": token_id}} if token_id else {"sale_history": {}}
return self.client.wasm.contract_query(self.config.nft_contract_addr, msg) or []
async def get_creator_analytics(self, creator: str) -> dict:
earnings = await self.get_royalty_earnings(creator)
total_volume = sum(int(e.get("amount", 0)) for e in earnings)
sale_count = len(earnings)
return {
"creator": creator, "total_earnings": total_volume,
"total_earnings_display": f"{total_volume / 1_000_000:,.2f} MSG",
"sale_count": sale_count, "average_earning": total_volume // sale_count if sale_count > 0 else 0,
"recent_earnings": earnings[-10:] if earnings else [],
}
7.4 收入分享
class RevenueSharing:
def __init__(self, config: AgentNFTConfig):
self.config = config
self.wallet = config.get_wallet()
self.client = config.get_client()
async def distribute_share(self, total_amount: int, shares: list[tuple[str, int]], denom: str = MSG_DENOM) -> dict:
total_bps = sum(bps for _, bps in shares)
if total_bps != 10000:
raise ValueError(f"Basis points must sum to 10000, got {total_bps}")
execute_msg = MsgExecuteContract(
sender=self.wallet.key.acc_address, contract=self.config.nft_contract_addr,
msg={"distribute": {}}, funds=[{"denom": denom, "amount": str(total_amount)}],
)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {
"action": "distribute", "total_amount": total_amount,
"total_display": f"{total_amount / 1_000_000:.2f} MSG",
"recipients": [{"address": addr, "bps": bps, "amount": total_amount * bps // 10000, "display": f"{total_amount * bps // 10000 / 1_000_000:.2f} MSG"} for addr, bps in shares],
"txhash": result.txhash,
}
async def setup_revenue_split(self, token_id: str, shares: list[tuple[str, int]]) -> dict:
msg = {"set_revenue_split": {"token_id": token_id, "shares": [{"address": addr, "basis_points": bps} for addr, bps in shares]}}
execute_msg = MsgExecuteContract(sender=self.wallet.key.acc_address, contract=self.config.nft_contract_addr, msg=msg)
simulated = self.client.estimate_fee(msgs=[execute_msg], wallet=self.wallet)
fee = Fee(amount=Coins.from_str(self.config.gas_price), gas=simulated[0].gas_limit)
tx = self.wallet.create_and_sign_tx(msgs=[execute_msg], fee=fee)
result = self.client.broadcast_tx(tx)
return {"action": "setup_revenue_split", "token_id": token_id, "shares": shares, "txhash": result.txhash}
8. 完整示例：Agent NFT Collection 启动
8.1 完整启动脚本
#!/usr/bin/env python3
\"\"\"
完整示例：AI Agent NFT Collection 启动脚本
本脚本演示一个 Agent 如何：
1. 实例化 NFT 合约
2. 铸造多种类型 NFT
3. 配置版税
4. 在市场挂单
5. 设置门控访问
\"\"\"
import asyncio
import json
import logging
import os
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("NFT-Launch")
AGENT_MNEMONIC = os.getenv("AGENT_MNEMONIC", "word1 word2 ... word24")
AGENT_ADDR = "msg1agent..."
USER_ADDR_1 = "msg1user1..."
USER_ADDR_2 = "msg1user2..."
WASM_CODE_ID = 123
async def main():
config = AgentNFTConfig(mnemonic=AGENT_MNEMONIC)
wallet = config.get_wallet()
# ============================================================
# Step 1: 实例化 NFT 合约
# ============================================================
logger.info("=== Step 1: 实例化 NFT 合约 ===")
init_msg = {
"name": "AI Agent Achievement Collection",
"symbol": "AGENTNFT",
"minter": AGENT_ADDR,
"agent_did": f"did:msg:{AGENT_ADDR}",
"nft_type": "achievement",
"description": "AI Agent 成就与会员 NFT 集合",
"image": "https://nft.msgchain.org/collections/agent-achievements.png",
"external_link": "https://msgchain.org/agents/nft",
"royalty_info": {"creator": AGENT_ADDR, "royalty_basis_points": 500},
}
instantiate_msg = MsgInstantiateContract(
sender=AGENT_ADDR, code_id=WASM_CODE_ID, msg=init_msg, funds=[], label="AI-Agent-NFT-Collection",
)
simulated = config.get_client().estimate_fee(msgs=[instantiate_msg], wallet=wallet)
fee = Fee(amount=Coins.from_str(config.gas_price), gas=simulated[0].gas_limit)
tx = wallet.create_and_sign_tx(msgs=[instantiate_msg], fee=fee)
result = config.get_client().broadcast_tx(tx)
logger.info(f"Contract deployed TX: {result.txhash}")
nft_contract_addr = "msg1nftcontract..." # parse from events
config.nft_contract_addr = nft_contract_addr
logger.info(f"NFT Contract: {nft_contract_addr}")
# ============================================================
# Step 2: 初始化服务
# ============================================================
logger.info("=== Step 2: 初始化服务 ===")
nft_service = AgentNFTService(config)
marketplace = NFTMarketplace(config, marketplace_addr="msg1marketplace...")
gated_access = NFTGatedAccess(config)
reputation = ReputationNFT(config)
# ============================================================
# Step 3: 铸造信誉徽章
# ============================================================
logger.info("=== Step 3: 铸造信誉徽章 ===")
result = await nft_service.mint_reputation_badge(
user_addr=USER_ADDR_1, total_score=85,
dimensions=[
{"name": "trading", "score": 90, "max_score": 100},
{"name": "contribution", "score": 80, "max_score": 100},
{"name": "governance", "score": 75, "max_score": 100},
],
)
logger.info(f"User1 reputation badge: {result['token_id']}")
result = await nft_service.mint_reputation_badge(user_addr=USER_ADDR_2, total_score=45)
logger.info(f"User2 reputation badge: {result['token_id']}")
# ============================================================
# Step 4: 铸造成就徽章
# ============================================================
logger.info("=== Step 4: 铸造成就徽章 ===")
achievements = [
(USER_ADDR_1, "first_transaction", "首次交易达成", 1),
(USER_ADDR_1, "volume_milestone", "累计 1000 笔交易", 3),
(USER_ADDR_2, "first_transaction", "首次交易达成", 1),
(USER_ADDR_2, "referral_star", "推荐 5 位新用户", 2),
]
for user, ach_type, title, level in achievements:
result = await nft_service.mint_achievement_badge(
user_addr=user, achievement_type=ach_type, title=title,
description=f"成就: {title}", level=level,
)
logger.info(f"Achievement minted: {result['title']} -> {result['token_id']}")
# ============================================================
# Step 5: 铸造会员通行证
# ============================================================
logger.info("=== Step 5: 铸造会员通行证 ===")
result = await nft_service.mint_membership_pass(
user_addr=USER_ADDR_1, tier="gold", duration_days=180,
benefits=["无限制 API 调用", "专属支持", "实时数据分析"],
)
logger.info(f"Gold membership for user1: {result['token_id']}, expires: {result['valid_until']}")
result = await nft_service.mint_membership_pass(
user_addr=USER_ADDR_2, tier="bronze", duration_days=30,
)
logger.info(f"Bronze membership for user2: {result['token_id']}")
# ============================================================
# Step 6: 铸造贡献证明
# ============================================================
logger.info("=== Step 6: 铸造贡献证明 ===")
result = await nft_service.mint_contribution_nft(
contribution_id="PR-42", contributor=USER_ADDR_1,
contribution_type="code", description="Added cross-chain NFT bridge",
amount=500, verifiers=[AGENT_ADDR],
)
logger.info(f"Contribution NFT: {result['token_id']}")
# ============================================================
# Step 7: 铸造访问通行证
# ============================================================
logger.info("=== Step 7: 铸造访问通行证 ===")
result = await nft_service.mint_access_pass(
user_addr=USER_ADDR_1,
permissions=["premium_query", "data_export", "batch_process"],
duration_days=90, max_uses=10000,
)
logger.info(f"Access pass for user1: {result['token_id']}")
# ============================================================
# Step 8: 市场挂单
# ============================================================
logger.info("=== Step 8: 市场挂单 ===")
# 用户2 挂单出售其青铜会员
result = await marketplace.list_nft(
token_id="<user2-membership-token-id>",
price=50_000_000, # 50 MSG
)
logger.info(f"Listed on marketplace: {result['txhash']}")
# ============================================================
# Step 9: 门控访问验证
# ============================================================
logger.info("=== Step 9: 门控访问验证 ===")
for user, feature in [(USER_ADDR_1, "premium_query"), (USER_ADDR_1, "admin_panel"), (USER_ADDR_2, "premium_query")]:
access = await gated_access.check_access(user, feature)
logger.info(f"User {user[:12]} -> {feature}: {'ALLOWED' if access else 'DENIED'}")
# ============================================================
# Step 10: 信誉系统集成
# ============================================================
logger.info("=== Step 10: 信誉系统集成 ===")
result = await reputation.issue_reputation_nft(
user=USER_ADDR_1,
categories_scores={
"trading": 95, "contribution": 88,
"governance": 70, "community": 85, "longevity": 60,
},
)
logger.info(f"Reputation issued: {result['action']}, level={result['level']}")
profile = await reputation.get_user_reputation_profile(USER_ADDR_1)
logger.info(f"User1 composite score: {profile['composite_score']}")
# 验证成就
verified = await reputation.verify_achievement(USER_ADDR_1, "first_transaction")
logger.info(f"User1 has first_transaction: {verified}")
# ============================================================
# Summary
# ============================================================
logger.info("=== Collection Launch Complete ===")
logger.info(f"Contract: {nft_contract_addr}")
logger.info(f"Minted: achievements={len(achievements)}, memberships=2, contributions=1, passes=1")
if __name__ == "__main__":
asyncio.run(main())
本文档为 MSG Chain AI Agent NFT 铸造与数字权益的完整指南。 涵盖了从合约开发、NFT 铸造、市场交易、门控访问到信誉系统集成的全链路。所有合约基于 CosmWasm CW721 标准，并针对 Agent 场景做了专门扩展。
关键词:
msg-chain-1msgCW721AI AgentNFTCosmWasmReputationGated AccessRoyalty