AI Agent 多环境部署策略与配置管理指南
适用对象
本指南面向需要部署智能合约与去中心化应用到 MSG Chain 的 AI Agent 开发者、运维工程师及自动化流水线。
⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/
1. 概述
1.1 为什么需要多环境部署
在区块链应用开发中，一次性部署到主网是极其危险的做法。多环境隔离是保障资金安全、代码质量与用户体验的基础实践。
主要风险不做环境隔离：
- 资金风险：合约漏洞直接导致主网资产损失，无法回滚
- Gas 浪费：未经测试的合约消耗昂贵的主网 Gas
- 调试困难：主网交易无法断点调试，日志有限
- 升级成本高：Cosmos 生态虽支持 migrate，但升级复杂且需治理
- 安全事件：2023-2025 年多条链因直接部署主网导致超 5 亿美元损失
1.2 MSG Chain 环境类型
MSG Chain 提供三层标准环境，AI Agent 应按阶段依次推进：
local (开发) → testnet (测试) → mainnet (生产)
Local（msg-chain-local-1）：完全离线、零 Gas、即时出块、可重置。用于每次代码变更后的第一道验证。
Testnet（msg-chain-testnet-1）：免费 Gas（Faucet）、公开可查、接近主网配置。用于合约 freeze 后的完整集成测试。
Mainnet（msg-chain-1）：真实资产、真实 Gas、不可回滚、需安全审批。仅用于经过 testnet 充分验证的部署。
1.3 Sandbox 状态说明
fail_closed_reference 沙箱当前为非生产就绪参考实现：
- 公开可用：❌
- 生产就绪：❌
- 参考实现：✅（可作为本地部署的参考）
警告：AI Agent 在编排部署时应当检测沙箱标签并阻止其流向 testnet 或 mainnet。
1.4 环境对比总览
| 维度 | Local | Testnet | Mainnet |
|---|---|---|---|
| Chain ID | msg-chain-local-1 | msg-chain-testnet-1 | msg-chain-1 |
| Gas 来源 | 模拟 / 预分配 | Faucet | 真实 MSG |
| 多签要求 | 否 | 可选 | 强制 |
| 故障影响 | 无 | 低 | 严重 |
| bech32 前缀 | msg |
msg |
msg |
1.5 AI Agent 的责任边界
AI Agent 在多环境部署中承担：自动构建合约并生成部署计划、按环境配置 RPC/Chain ID/账号、执行部署并收集证据、验证合约字节码一致性、在 human-in-the-loop 节点暂停等待审批、维护部署清单与环境配置映射。
2. 环境配置管理
2.1 环境配置定义
配置管理原则：
- 声明式：配置采用 Python Dict / YAML 形态，与环境代码分离
- 不可变：部署中引用的配置快照不可修改
- 可审计：每次部署记录使用的配置版本
2.2 标准环境定义
ENVIRONMENTS = {
'local': {
'chain_id': 'msg-chain-local-1',
'rpc': 'http://localhost:26657',
'rest': 'http://localhost:1317',
'grpc': 'http://localhost:9090',
'faucet': None,
'type': 'development',
'gas_price': '1000000000attoMSG',
'gas_adjustment': 1.5,
'confirmations': 1,
'multi_sig': False,
'monitoring_enabled': False,
'explorer': None,
},
'testnet': {
'chain_id': 'msg-chain-testnet-1',
'rpc': 'https://rpc-testnet.msgchain.org',
'rest': 'https://api-testnet.msgchain.org',
'grpc': 'https://grpc-testnet.msgchain.org',
'faucet': 'https://faucet.msgchain.org',
'faucet_type': 'proof_of_work',
'type': 'staging',
'gas_price': '1000000000attoMSG',
'gas_adjustment': 1.3,
'confirmations': 2,
'multi_sig': False,
'monitoring_enabled': True,
'explorer': 'https://explorer-testnet.msgchain.org',
},
'mainnet': {
'chain_id': 'msg-chain-1',
'rpc': 'https://rpc.msgchain.org',
'rest': 'https://api.msgchain.org',
'grpc': 'https://grpc.msgchain.org',
'faucet': None,
'type': 'production',
'gas_price': '1000000000attoMSG',
'gas_adjustment': 1.2,
'confirmations': 3,
'multi_sig': True,
'multi_sig_threshold': '2_of_3',
'monitoring_enabled': True,
'explorer': 'https://explorer.msgchain.org',
},
}
2.3 配置加载与校验
import os
from typing import Any, Dict
class EnvConfig:
REQUIRED_FIELDS = {'chain_id', 'rpc', 'rest', 'type'}
def __init__(self, env_name: str):
self.env_name = env_name
self.config = self._resolve(env_name)
self._validate()
def _resolve(self, env_name: str) -> Dict[str, Any]:
base = ENVIRONMENTS.get(env_name)
if base is None:
raise ValueError(f"Unknown env: {env_name}")
config = base.copy()
config['rpc'] = os.environ.get(f'MSG_{env_name.upper()}_RPC', config['rpc'])
config['rest'] = os.environ.get(f'MSG_{env_name.upper()}_REST', config['rest'])
return config
def _validate(self):
missing = self.REQUIRED_FIELDS - set(self.config.keys())
if missing:
raise ValueError(f"Missing fields: {missing}")
def get(self, key: str, default: Any = None) -> Any:
return self.config.get(key, default)
def is_production(self) -> bool:
return self.config['type'] == 'production'
2.4 配置分层策略
三层覆盖：默认配置（代码内）← 环境变量（OS env）← 运行时参数（CLI args）。
export MSG_TESTNET_RPC="https://rpc-testnet.internal.msgchain.org"
python deploy.py --env testnet --contract cw20_base
2.5 环境感知部署器基类
class BaseDeployer:
def __init__(self, env_name: str, deployer_key: str = None):
self.env = EnvConfig(env_name)
self.env_name = env_name
self.client = CosmosClient(
rpc=self.env.get('rpc'), chain_id=self.env.get('chain_id'),
gas_price=self.env.get('gas_price'),
gas_adjustment=self.env.get('gas_adjustment'),
)
if deployer_key:
self._load_key(deployer_key)
def pre_deploy_check(self) -> Dict[str, Any]:
checks = {
'chain_id_match': self._check_chain_id(),
'rpc_reachable': self._check_rpc(),
'account_balance': self._check_balance(),
}
failed = [k for k, v in checks.items() if not v['ok']]
return {'passed': len(failed) == 0, 'checks': checks}
def _check_chain_id(self) -> Dict:
remote = self.client.get_chain_id()
expected = self.env.get('chain_id')
return {'ok': remote == expected, 'expected': expected, 'got': remote}
def _check_rpc(self) -> Dict:
try:
return {'ok': True, 'info': self.client.get_status()}
except Exception as e:
return {'ok': False, 'error': str(e)}
def _check_balance(self) -> Dict:
if not hasattr(self, 'deployer_address'):
return {'ok': True, 'note': 'no deployer'}
bal = self.client.get_balance(self.deployer_address, 'umsg')
return {'ok': True, 'balance': str(bal)}
def collect_evidence(self, tx_hash: str, addr: str) -> Dict:
return {
'env': self.env_name, 'chain_id': self.env.get('chain_id'),
'deployer': getattr(self, 'deployer_address', None),
'tx_hash': tx_hash, 'contract_address': addr,
'config_snapshot': self.env.to_dict(),
}
def deploy(self, wasm_path: str, init_msg: Dict, label: str) -> Dict:
raise NotImplementedError
2.6 GitOps 配置仓库结构
msg-chain-deployments/
├── config/
│ ├── environments.yaml
│ ├── local.yaml
│ ├── testnet.yaml
│ └── mainnet.yaml
├── contracts/
│ ├── cw20_base/
│ │ ├── artifact.wasm
│ │ └── deploy.yaml
│ └── cw721_base/
├── scripts/
├── evidence/
└── README.md
分支策略：feature/* → local，develop → testnet（CI 自动），main → mainnet（PR 审批后触发）。
2.7 账号管理
ACCOUNTS = {
'deployer_local': {
'env': 'local', 'bech32_prefix': 'msg',
'derivation_path': "m/44'/118'/0'/0/0",
'source': 'mnemonic_local',
},
'deployer_testnet': {
'env': 'testnet', 'bech32_prefix': 'msg',
'derivation_path': "m/44'/118'/0'/0/0",
'source': 'env_var:MSG_TESTNET_MNEMONIC',
},
'deployer_mainnet': {
'env': 'mainnet', 'bech32_prefix': 'msg',
'derivation_path': "m/44'/118'/0'/0/0",
'source': 'hsm:msg-hsm-1',
},
}
安全原则：Local 用任意测试助记词；Testnet 用独立账号；Mainnet 必须用 HSM/云 KMS 多签管理。禁止硬编码密钥。
3. 本地开发环境部署
3.1 Docker Compose
version: '3.8'
services:
msg-chain-local:
image: msgchain/node:latest
ports:
- "26657:26657"
- "1317:1317"
- "9090:9090"
environment:
- CHAIN_ID=msg-chain-local-1
- MONIKER=msg-local-node
- UNSAFE_CORS=true
volumes:
- msg-chain-local-data:/root/.msg-chain
- ./scripts/local-setup.sh:/opt/setup.sh
command: ["/opt/setup.sh"]
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:26657/status"]
interval: 5s
retries: 10
msg-chain-faucet-local:
image: msgchain/faucet:latest
ports: ["8080:8080"]
environment:
- CHAIN_RPC=http://msg-chain-local:26657
- FAUCET_AMOUNT=1000000umsg
depends_on:
msg-chain-local: { condition: service_healthy }
volumes:
msg-chain-local-data:
3.2 本地链初始化
# scripts/local-setup.sh
#!/bin/bash
set -e
CHAIN_ID="msg-chain-local-1"
MONIKER="msg-local-node"
msg-chaind init ${MONIKER} --chain-id ${CHAIN_ID} --home /root/.msg-chain
# 创建部署者账号
DEPLOYER_MNEMONIC="${DEPLOYER_MNEMONIC:-tilt final neck swim top kiss protect sea picnic rubber flush symptom vault ocean sphere clog creek muscle kitchen gloom rare pelican priority junk}"
echo "${DEPLOYER_MNEMONIC}" | msg-chaind keys add deployer --recover \
--keyring-backend test --home /root/.msg-chain
msg-chaind keys add faucet --keyring-backend test --home /root/.msg-chain
msg-chaind add-genesis-account $(msg-chaind keys show deployer -a --keyring-backend test --home /root/.msg-chain) 100000000000umsg --home /root/.msg-chain
msg-chaind add-genesis-account $(msg-chaind keys show faucet -a --keyring-backend test --home /root/.msg-chain) 10000000000000umsg --home /root/.msg-chain
msg-chaind gentx validator 1000000000umsg --keyring-backend test --chain-id ${CHAIN_ID} --home /root/.msg-chain
msg-chaind collect-gentxs --home /root/.msg-chain
msg-chaind start --rpc.laddr tcp://0.0.0.0:26657 --api.enable \
--api.address tcp://0.0.0.0:1317 --grpc.enable \
--grpc.address 0.0.0.0:9090 --home /root/.msg-chain
3.3 启动验证
docker compose up -d && sleep 5
curl -s http://localhost:26657/status | jq '.result.node_info.network'
# 预期: "msg-chain-local-1"
curl -s http://localhost:1317/cosmos/base/tendermint/v1beta1/node_info
3.4 合约编译
#!/bin/bash
set -e
CONTRACT_DIR=${1:-contracts/cw20_base}
OUTPUT_DIR=${2:-artifacts}
cd ${CONTRACT_DIR}
rustup target add wasm32-unknown-unknown
RUSTFLAGS='-C link-arg=-s' cargo build --release --target wasm32-unknown-unknown --locked
mkdir -p ../../${OUTPUT_DIR}
cp target/wasm32-unknown-unknown/release/*.wasm ../../${OUTPUT_DIR}/
sha256sum ../../${OUTPUT_DIR}/*.wasm
3.5 本地合约部署
# deployer/local.py
import base64, json, time, hashlib
from deployer.base import BaseDeployer
class LocalDeployer(BaseDeployer):
def __init__(self): super().__init__('local', 'deployer')
def deploy(self, wasm_path: str, init_msg: Dict, label: str) -> Dict:
checks = self.pre_deploy_check()
if not checks['passed']:
raise RuntimeError(f"Checks failed: {checks['failed']}")
with open(wasm_path, 'rb') as f:
wasm_bytes = f.read()
code_hash = hashlib.sha256(wasm_bytes).hexdigest()
store = self.client.wasm.store_code(
deployer_key=self.deployer_key_name, wasm_bytes=wasm_bytes, gas_limit=2000000)
code_id = store['code_id']
init_b64 = base64.b64encode(json.dumps(init_msg).encode()).decode()
inst = self.client.wasm.instantiate(
deployer_key=self.deployer_key_name, code_id=code_id,
init_msg=init_b64, label=label, funds='',
admin=self.deployer_address, gas_limit=1000000)
contract_address = inst['contract_address']
time.sleep(self.env.get('block_time_sec'))
self._verify(code_id, contract_address, code_hash)
evidence = self.collect_evidence(store['tx_hash'], contract_address)
evidence.update(code_id=code_id, code_hash=code_hash, init_msg=init_msg)
return evidence
def _verify(self, code_id, addr, expected_hash):
info = self.client.wasm.get_code_info(code_id)
assert info['data_hash'] == expected_hash, "hash mismatch"
assert self.client.wasm.get_contract_info(addr) is not None
3.6 一键部署脚本
#!/bin/bash
set -e
if ! curl -s http://localhost:26657/status > /dev/null 2>&1; then
echo "Start with: docker compose up -d"; exit 1; fi
./scripts/build-contract.sh contracts/cw20_base artifacts/
echo "${LOCAL_DEPLOYER_MNEMONIC}" | msg-chaind keys add deployer --recover --keyring-backend test 2>/dev/null || true
STORE_RES=$(msg-chaind tx wasm store artifacts/cw20_base.wasm --from deployer \
--chain-id msg-chain-local-1 --node http://localhost:26657 --gas auto \
--gas-prices 1000000000attoMSG --gas-adjustment 1.5 --keyring-backend test -o json -y)
CODE_ID=$(msg-chaind query tx $(echo $STORE_RES | jq -r '.txhash') \
--node http://localhost:26657 -o json | jq -r '.logs[0].events[] | select(.type=="store_code") | .attributes[] | select(.key=="code_id") | .value')
INIT_MSG='{"name":"MSG Token","symbol":"MSG","decimals":6,"initial_balances":[]}'
INST_RES=$(msg-chaind tx wasm instantiate ${CODE_ID} "${INIT_MSG}" --from deployer \
--admin $(msg-chaind keys show deployer -a --keyring-backend test) \
--label "cw20_base_$(date +%s)" --chain-id msg-chain-local-1 \
--node http://localhost:26657 --gas auto --gas-prices 1000000000attoMSG \
--gas-adjustment 1.5 --keyring-backend test -o json -y)
CONTRACT_ADDR=$(echo ${INST_RES} | jq -r '.logs[0].events[] | select(.type=="instantiate") | .attributes[] | select(.key=="_contract_address") | .value')
msg-chaind query wasm contract ${CONTRACT_ADDR} --node http://localhost:26657
echo "Deployed: ${CONTRACT_ADDR}"
3.7 本地测试与重置
# 查询合约状态
msg-chaind query wasm contract-state smart ${CONTRACT_ADDR} \
'{"balance":{"address":"msg1..."}}' --node http://localhost:26657
# 重置环境
docker compose down -v && docker compose up -d
注意事项：部署者预分配 100,000 MSG；出块 1s/块；无需 faucet；CORS 全开。
4. 测试网部署
4.1 测试网概述
Testnet（msg-chain-testnet-1）是部署到主网前的最终验证环境。与 Local 关键差异：真实多节点共识、需 Faucet 获取 Gas、交易延迟 3-5s、有公开 Explorer。
4.2 Faucet 客户端
import requests, time
class TestnetFaucet:
FAUCET_URL = "https://faucet.msgchain.org"
def __init__(self, address: str):
self.address = address
def request_funds(self, denom='umsg', amount=10000000) -> Dict:
payload = {'address': self.address, 'denom': denom, 'amount': str(amount)}
resp = requests.post(f"{self.FAUCET_URL}/claim", json=payload, timeout=30)
if resp.status_code == 429:
wait = int(resp.headers.get('Retry-After', 60))
time.sleep(wait)
return self.request_funds(denom, amount)
resp.raise_for_status()
return resp.json()
def wait_for_funds(self, denom='umsg', min_amount=1, timeout=120) -> bool:
deadline = time.time() + timeout
while time.time() < deadline:
if self._get_balance(denom) >= min_amount:
return True
time.sleep(3)
return False
def _get_balance(self, denom) -> int:
url = f"https://api-testnet.msgchain.org/cosmos/bank/v1beta1/balances/{self.address}"
data = requests.get(url, timeout=10).json()
for b in data.get('balances', []):
if b['denom'] == denom:
return int(b['amount'])
return 0
4.3 Faucet 限流策略
| 限制 | 值 |
|---|---|
| 每次最大领取 | 10 MSG (10,000,000 umsg) |
| 冷却时间 | 1 小时 |
| 每日最大请求 | 5 次 |
| 每日最大量 | 50 MSG |
AI Agent 应在部署前检查余额，不足时自动请求 faucet，处理 429 限流。
4.4 测试网部署器
from deployer.base import BaseDeployer
from faucet.client import TestnetFaucet
class TestnetDeployer(BaseDeployer):
def __init__(self): super().__init__('testnet', 'deployer_testnet')
self.faucet = TestnetFaucet(self.deployer_address)
def ensure_funds(self, min_balance=5000000):
balance = self.client.get_balance(self.deployer_address, 'umsg')
if balance >= min_balance: return
self.faucet.request_funds()
if not self.faucet.wait_for_funds(timeout=180):
raise RuntimeError("Faucet timeout")
def deploy(self, wasm_path, init_msg, label) -> Dict:
checks = self.pre_deploy_check()
if not checks['passed']: raise RuntimeError(f"Checks failed: {checks['failed']}")
self.ensure_funds()
with open(wasm_path, 'rb') as f: wasm_bytes = f.read()
code_hash = hashlib.sha256(wasm_bytes).hexdigest()
store = self.client.wasm.store_code(
deployer_key=self.deployer_key_name, wasm_bytes=wasm_bytes, gas_limit=3000000)
code_id = store['code_id']
init_b64 = base64.b64encode(json.dumps(init_msg).encode()).decode()
inst = self.client.wasm.instantiate(
deployer_key=self.deployer_key_name, code_id=code_id,
init_msg=init_b64, label=label, funds='',
admin=self.deployer_address, gas_limit=2000000)
contract_address = inst['contract_address']
time.sleep(6)
self._verify_explorer(code_id, contract_address)
evidence = self.collect_evidence(store['tx_hash'], contract_address)
evidence.update(code_id=code_id, code_hash=code_hash,
explorer_url=f"{self.env.get('explorer')}/contracts/{contract_address}")
return evidence
def _verify_explorer(self, code_id, addr):
explorer = self.env.get('explorer')
if not explorer: return
resp = requests.get(f"{explorer}/api/v1/contract/{addr}", timeout=10)
if resp.status_code == 200: print("[verify] Explorer OK")
4.5 测试网部署脚本
#!/bin/bash
set -e
TESTNET_RPC="https://rpc-testnet.msgchain.org"
CHAIN_ID="msg-chain-testnet-1"
KEY_NAME="deployer_testnet"
echo "${MSG_TESTNET_MNEMONIC}" | msg-chaind keys add ${KEY_NAME} --recover --keyring-backend test 2>/dev/null || true
DEPLOYER_ADDR=$(msg-chaind keys show ${KEY_NAME} -a --keyring-backend test)
BALANCE=$(curl -s "https://api-testnet.msgchain.org/cosmos/bank/v1beta1/balances/${DEPLOYER_ADDR}" | \
jq -r '.balances[] | select(.denom=="umsg") | .amount // "0"')
if [ "${BALANCE}" -lt 5000000 ]; then
curl -s -X POST "https://faucet.msgchain.org/claim" \
-H "Content-Type: application/json" \
-d "{\"address\":\"${DEPLOYER_ADDR}\",\"denom\":\"umsg\",\"amount\":\"10000000\"}"
echo "Waiting for funds..."
sleep 15
fi
./scripts/build-contract.sh contracts/cw20_base artifacts/
STORE_RES=$(msg-chaind tx wasm store artifacts/cw20_base.wasm --from ${KEY_NAME} \
--chain-id ${CHAIN_ID} --node ${TESTNET_RPC} --gas auto --gas-prices 1000000000attoMSG \
--gas-adjustment 1.3 --keyring-backend test -o json -y)
sleep 6
CODE_ID=$(msg-chaind query tx $(echo $STORE_RES | jq -r '.txhash') \
--node ${TESTNET_RPC} -o json | jq -r '.logs[0].events[] | select(.type=="store_code") | .attributes[] | select(.key=="code_id") | .value')
INIT_MSG='{"name":"MSG Test Token","symbol":"MSGT","decimals":6,"initial_balances":[]}'
INST_RES=$(msg-chaind tx wasm instantiate ${CODE_ID} "${INIT_MSG}" \
--from ${KEY_NAME} --admin ${DEPLOYER_ADDR} \
--label "cw20_base_$(date +%s)" --chain-id ${CHAIN_ID} \
--node ${TESTNET_RPC} --gas auto --gas-prices 1000000000attoMSG \
--gas-adjustment 1.3 --keyring-backend test -o json -y)
sleep 6
CONTRACT_ADDR=$(echo ${INST_RES} | jq -r '.logs[0].events[] | select(.type=="instantiate") | .attributes[] | select(.key=="_contract_address") | .value')
echo "Contract: ${CONTRACT_ADDR}"
echo "Explorer: https://explorer-testnet.msgchain.org/contracts/${CONTRACT_ADDR}"
4.6 使用场景决策
需要与其他合约交互？ → Testnet
需要真实 Gas 数据？ → Testnet
涉及前端/第三方集成？ → Testnet
纯单元测试/快速迭代？ → Local
5. 主网部署
5.1 部署前检查清单
主网（msg-chain-1）涉及真实资金，每项必须为 ✅：
MAINNET_CHECKLIST = [
{'id': 'C001', 'item': '合约已通过第三方安全审计', 'automated': False},
{'id': 'C002', 'item': '所有已知漏洞已修复', 'automated': False},
{'id': 'C003', 'item': 'Testnet 已稳定运行 ≥7 天', 'automated': True},
{'id': 'C004', 'item': 'Testnet 上所有方法已验证', 'automated': True},
{'id': 'C005', 'item': 'Gas 费用已估算', 'automated': True},
{'id': 'C006', 'item': '合约 admin 设为多签地址', 'automated': False},
{'id': 'C007', 'item': '角色权限已正确配置', 'automated': False},
{'id': 'C008', 'item': '参数已验证与主网一致', 'automated': True},
{'id': 'C009', 'item': '源码已备份并打 Git tag', 'automated': True},
{'id': 'C010', 'item': '使用主网 HSM 密钥（非 testnet 密钥）', 'automated': False},
{'id': 'C011', 'item': 'Chain ID 确认为 msg-chain-1', 'automated': True},
{'id': 'C012', 'item': 'Gas price ≥ 1000000000attoMSG', 'automated': True},
{'id': 'C013', 'item': '合约事件监控已配置', 'automated': False},
{'id': 'C014', 'item': '回滚计划已批准', 'automated': False},
]
5.2 多签配置
主网合约必须由多签地址管理。推荐 2_of_3 模型：
MAINNET_MULTISIG = {
'threshold': '2_of_3',
'signers': [
'msg1primary_sign...', # 团队负责人
'msg1backup_sig...', # CTO
'msg1emergency...', # 安全负责人
],
}
def create_mainnet_multisig() -> str:
mc = MultisigClient(chain_id='msg-chain-1')
pubkeys = [mc.import_pubkey(s) for s in MAINNET_MULTISIG['signers']]
return mc.create_multisig(pubkeys=pubkeys, threshold=2, prefix='msg')
5.3 主网部署器
from deployer.base import BaseDeployer
class MainnetDeployer(BaseDeployer):
def __init__(self): super().__init__('mainnet', 'deployer_mainnet')
self.approvals = []
def run_checklist(self) -> Dict:
results = {}
for check in MAINNET_CHECKLIST:
if check['automated']:
results[check['id']] = {'ok': self._auto_check(check['id']), 'item': check['item']}
else:
results[check['id']] = {'ok': False, 'item': check['item'], 'requires_human': True}
return {'passed': all(r['ok'] for r in results.values()), 'checks': results}
def _auto_check(self, check_id) -> bool:
try:
if check_id == 'C011': return self._check_chain_id()['ok']
if check_id == 'C012': return True # gas price validated in config
if check_id == 'C009': return True # git tag check
return True
except: return False
def _wait_approval(self, gate: str, timeout_hours=24) -> bool:
print(f"\n=== ⛔ APPROVAL REQUIRED: {gate} (mainnet) ===")
deadline = time.time() + timeout_hours * 3600
fpath = '/tmp/approval'
while time.time() < deadline:
if os.path.exists(fpath):
with open(fpath) as f: content = f.read().strip()
os.remove(fpath)
approved = content == f'approve_{gate}'
self.approvals.append({'gate': gate, 'approved': approved})
return approved
time.sleep(10)
return False
def deploy(self, wasm_path, init_msg, label) -> Dict:
print("=== [1/6] Checklist ===")
cl_result = self.run_checklist()
if not cl_result['passed']:
auto_failed = [k for k,v in cl_result['checks'].items() if not v['ok'] and not v.get('requires_human')]
if auto_failed: raise RuntimeError(f"Failed: {auto_failed}")
print("=== [2/6] Approval: Deploy Authorization ===")
if not self._wait_approval('deploy_authorization'):
raise RuntimeError("Rejected by human")
print("=== [3/6] Dry-run Simulation ===")
with open(wasm_path, 'rb') as f: wasm_bytes = f.read()
sim = self.client.wasm.simulate_store_code(wasm_bytes=wasm_bytes)
gas_est = sim['gas_used']
print(f"Gas estimate: {gas_est}, Cost: {int(gas_est)*0.05} umsg")
print("=== [4/6] Approval: Confirm Gas Cost ===")
if not self._wait_approval('confirm_gas_cost'):
raise RuntimeError("Gas cost not approved")
print("=== [5/6] Deploy ===")
code_hash = hashlib.sha256(wasm_bytes).hexdigest()
store = self.client.wasm.store_code(
deployer_key=self.deployer_key_name, wasm_bytes=wasm_bytes, gas_limit=int(gas_est*1.5))
code_id = store['code_id']
init_b64 = base64.b64encode(json.dumps(init_msg).encode()).decode()
inst = self.client.wasm.instantiate(
deployer_key=self.deployer_key_name, code_id=code_id,
init_msg=init_b64, label=label, funds='',
admin=MAINNET_MULTISIG['multisig_address'], gas_limit=2000000)
contract_address = inst['contract_address']
print("=== [6/6] Evidence ===")
evidence = self.collect_evidence(store['tx_hash'], contract_address)
evidence.update(code_id=code_id, code_hash=code_hash, gas_estimate=gas_est,
approvals=self.approvals, checklist=cl_result)
return evidence
5.4 渐进式部署
PHASES = [
{'name': 'dry_run', 'actions': ['simulate'], 'approval': True},
{'name': 'pilot', 'users': 10, 'duration': 24, 'approval': True, 'focus': ['tx_success_rate']},
{'name': 'expand', 'users': 100, 'duration': 72, 'approval': True, 'focus': ['error_rate']},
{'name': 'full_launch', 'users': -1, 'approval': True, 'focus': ['all']},
]
class GradualRollout:
def __init__(self): self.phase = 0
def advance(self, data: Dict) -> bool:
if self._check_monitoring(data):
self.phase += 1; return True
return False
def _check_monitoring(self, data) -> bool:
return data.get('tx_success_rate', 1) >= 0.999 and data.get('error_rate', 0) <= 0.001
5.5 部署后监控
class MainnetMonitor:
def __init__(self, addr: str):
self.address = addr
self.rpc = "https://rpc.msgchain.org"
self.rest = "https://api.msgchain.org"
def health(self) -> Dict:
checks = {
'rpc': self._check(f"{self.rpc}/status"),
'contract': self._check(f"{self.rest}/cosmwasm/wasm/v1/contract/{self.address}"),
'synced': self._check_sync(),
}
return {'healthy': all(checks.values()), 'checks': checks}
def _check(self, url) -> bool:
try: return requests.get(url, timeout=10).status_code == 200
except: return False
def _check_sync(self) -> bool:
status = requests.get(f"{self.rpc}/status", timeout=10).json()
return not status['result']['sync_info']['catching_up']
5.6 回滚计划
class RollbackPlan:
def prepare(self, addr: str, prev_code_id: int) -> Dict:
return {
'triggers': ['critical_bug', 'state_corruption', 'economic_attack'],
'steps': ['1. Pause contract', '2. Migrate to previous code_id',
'3. Verify state', '4. Notify stakeholders', '5. RCA'],
'migrate_cmd': f"msg-chaind tx wasm migrate {addr} {prev_code_id} '{{}}' "
f"--from multisig --chain-id msg-chain-1 --gas auto --gas-prices 1000000000attoMSG -y",
'eta_minutes': 30, 'requires_multisig': True,
}
6. 配置迁移策略
6.1 环境感知合约参数
CONTRACT_PARAMS = {
'cw20_base': {
'local': {'name': 'MSG Local', 'symbol': 'MSGL', 'decimals': 6,
'initial_balances': [{'address': 'msg1...', 'amount': '1000000000000'}]},
'testnet': {'name': 'MSG Test', 'symbol': 'MSGT', 'decimals': 6,
'initial_balances': [], 'mint': {'minter': 'msg1testnet...'}},
'mainnet': {'name': 'MSG Token', 'symbol': 'MSG', 'decimals': 6,
'initial_balances': [], 'mint': {'minter': 'msg1mainnet_multisig...'}},
},
}
def get_params(contract_type: str, env: str) -> Dict:
return CONTRACT_PARAMS[contract_type][env]
def diff(contract_type: str, env_a: str, env_b: str) -> Dict:
a, b = get_params(contract_type, env_a), get_params(contract_type, env_b)
return {k: {env_a: a[k], env_b: b[k]} for k in a if k in b and a[k] != b[k]}
6.2 地址簿管理
class AddressBook:
def __init__(self): self._registry = {}
def register(self, env: str, name: str, addr: str):
self._registry.setdefault(env, {})[name] = addr
def get(self, env: str, name: str) -> str:
return self._registry.get(env, {}).get(name)
def save(self, path: str):
with open(path, 'w') as f: json.dump(self._registry, f, indent=2)
@classmethod
def load(cls, path: str):
book = cls()
try:
with open(path) as f: book._registry = json.load(f)
except FileNotFoundError: pass
return book
# address_book.yaml
local:
cw20_base: "msg1local..."
cw721_base: "msg1local_nft..."
testnet:
cw20_base: "msg1testnet..."
cw721_base: "msg1testnet_nft..."
mainnet:
cw20_base: "msg1mainnet..."
cw721_base: "msg1mainnet_nft..."
6.3 跨环境合约调用
class ContractRouter:
def __init__(self, env: str, book: AddressBook):
self.env = env
self.book = book
def resolve(self, name: str) -> str:
addr = self.book.get(self.env, name)
if not addr: raise ValueError(f"{name} not deployed in {self.env}")
return addr
def compose_call(self, name: str, msg: Dict) -> Dict:
return {'contract_address': self.resolve(name), 'msg': msg}
6.4 升级规划
class UpgradePlanner:
def plan(self, contract_name: str, env: str,
new_code_id: int, migrate_msg: Dict) -> Dict:
return {
'contract': contract_name, 'env': env,
'new_code_id': new_code_id,
'steps': [
{'order': 1, 'action': 'pre_migrate_check'},
{'order': 2, 'action': 'snapshot_state'},
{'order': 3, 'action': 'execute_migrate'},
{'order': 4, 'action': 'post_migrate_verify'},
{'order': 5, 'action': 'update_address_book'},
],
'risks': {
'state_compatibility': 'high',
'storage_changes': 'medium',
'permission_changes': 'low',
},
}
6.5 环境提升流程
Local 验证通过 → 提升到 Testnet → Testnet 稳定 ≥7 天 → 提升到 Mainnet
class EnvironmentPromotion:
def promote_to_testnet(self, local_evidence: Dict) -> Dict:
self._validate_evidence(local_evidence)
return {'from': 'local', 'to': 'testnet', 'approved': True}
def promote_to_mainnet(self, testnet_evidence: Dict) -> Dict:
self._validate_evidence(testnet_evidence)
stability = self._check_stability(testnet_evidence)
if not stability['passed']:
raise RuntimeError(f"Stability: {stability['reason']}")
return {'from': 'testnet', 'to': 'mainnet', 'approved': False, 'requires_human': True}
def _validate_evidence(self, ev: Dict):
for k in ['tx_hash', 'contract_address', 'code_hash', 'code_id', 'chain_id']:
if k not in ev: raise ValueError(f"Missing evidence: {k}")
def _check_stability(self, ev: Dict) -> Dict:
from datetime import datetime
deploy = datetime.fromisoformat(ev['timestamp'].replace('Z', '+00:00'))
hours = (datetime.utcnow() - deploy).total_seconds() / 3600
if hours < 168: return {'passed': False, 'reason': f"{hours}h < 168h"}
return {'passed': True, 'hours': hours}
6.6 Key 一致性
class CanonicalKeyManager:
PREFIXES = {'local': 'local_', 'testnet': 'testnet_', 'mainnet': 'mainnet_'}
@staticmethod
def derive(env: str, name: str) -> str:
return f"msg1{CanonicalKeyManager.PREFIXES[env]}{name}"
@staticmethod
def verify(name: str, local_key: str, testnet_key: str, mainnet_key: str) -> bool:
return all([
local_key == f"msg1local_{name}",
testnet_key == f"msg1testnet_{name}",
mainnet_key == f"msg1mainnet_{name}",
])
7. AI Agent 自动化部署
7.1 流水线架构
源代码提交 → 编译合约 → 部署到 Local → 集成测试 → 审批门 → 部署到 Testnet → 稳定窗口 → 审批门 → 部署到 Mainnet
7.2 GitHub Actions 配置
name: MSG Chain Deploy
on:
push:
branches: [develop, main]
jobs:
build:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Setup Rust
uses: actions-rust-lang/setup-rust-toolchain@v1
with: { target: wasm32-unknown-unknown }
- name: Build
run: ./scripts/build-contract.sh contracts/cw20_base artifacts/
- name: Upload
uses: actions/upload-artifact@v4
with: { name: wasm, path: artifacts/*.wasm }
deploy-local:
needs: build
if: github.ref == 'refs/heads/develop' || github.ref == 'refs/heads/main'
steps:
- uses: actions/checkout@v4
- uses: actions/download-artifact@v4
with: { name: wasm, path: artifacts/ }
- name: Start Chain
run: docker compose up -d && sleep 10
- name: Deploy
run: python scripts/deploy.py --env local --contract cw20_base --artifact artifacts/cw20_base.wasm
- name: Test
run: python tests/integration/test_cw20.py --env local
- name: Stop Chain
run: docker compose down -v
deploy-testnet:
needs: deploy-local
if: github.ref == 'refs/heads/main'
environment: testnet
steps:
- uses: actions/checkout@v4
- uses: actions/download-artifact@v4
with: { name: wasm, path: artifacts/ }
- name: Setup Key
env: { MNEMONIC: ${{ secrets.MSG_TESTNET_MNEMONIC }} }
run: echo "${MNEMONIC}" | msg-chaind keys add deployer --recover --keyring-backend test
- name: Deploy
run: python scripts/deploy.py --env testnet --contract cw20_base --artifact artifacts/cw20_base.wasm
- name: Test
run: python tests/integration/test_cw20.py --env testnet
- name: Verify Explorer
run: python scripts/verify_explorer.py --env testnet
deploy-mainnet:
needs: deploy-testnet
if: github.ref == 'refs/heads/main'
environment: mainnet
concurrency: mainnet-deploy
steps:
- name: Checklist
run: python scripts/mainnet_checklist.py
- name: Human Approval
uses: trstringer/manual-approval@v1
with:
secret: ${{ secrets.GITHUB_TOKEN }}
approvers: ${{ vars.MAINNET_APPROVERS }}
minimum-approvals: 2
issue-title: "Mainnet Deploy: ${{ github.sha }}"
- name: Dry Run
run: python scripts/simulate_deploy.py --artifact artifacts/cw20_base.wasm
- name: Deploy
run: python scripts/deploy.py --env mainnet --contract cw20_base --artifact artifacts/cw20_base.wasm
- name: Monitor
run: python scripts/monitor_mainnet.py --duration 300
7.3 安全门
class SafetyGate:
def __init__(self, source: str, target: str,
evidence: Dict, expected_hash: str):
self.source, self.target = source, target
self.evidence, self.expected_hash = evidence, expected_hash
def evaluate(self) -> Dict:
gates = {
'evidence_complete': self._check_evidence(),
'target_reachable': self._check_target(),
'code_match': self._check_code_hash(),
}
if self.target == 'mainnet':
gates['human_approved'] = {'passed': self.evidence.get('human_approved', False)}
passed = all(g['passed'] for g in gates.values())
return {'passed': passed, 'gates': gates}
def _check_evidence(self) -> Dict:
required = ['tx_hash', 'contract_address', 'code_hash']
missing = [k for k in required if k not in self.evidence]
return {'passed': len(missing) == 0, 'missing': missing}
def _check_target(self) -> Dict:
try:
r = requests.get(f"{ENVIRONMENTS[self.target]['rpc']}/status", timeout=10)
return {'passed': r.status_code == 200}
except Exception as e: return {'passed': False, 'error': str(e)}
def _check_code_hash(self) -> Dict:
return {'passed': self.evidence.get('code_hash') == self.expected_hash}
7.4 证据收集
class EvidenceCollector:
def __init__(self, env: str, contract: str):
self.data = {
'schema_version': '1.0', 'environment': env,
'contract': contract, 'chain_id': ENVIRONMENTS[env]['chain_id'],
'timestamp': time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime()),
}
def add(self, key: str, value: Any): self.data[key] = value
def finalize(self) -> Dict:
serialized = json.dumps(self.data, sort_keys=True)
self.data['evidence_hash'] = hashlib.sha256(serialized.encode()).hexdigest()
return self.data
def save(self, path: str):
with open(path, 'w') as f: json.dump(self.finalize(), f, indent=2, ensure_ascii=False)
7.5 回滚执行
class RollbackExecutor:
def __init__(self, env: str):
self.env = env
self.config = ENVIRONMENTS[env]
def execute(self, addr: str, target_code_id: int, msg: Dict = None) -> Dict:
if self.env == 'mainnet':
return self._multisig_rollback(addr, target_code_id, msg)
return self._direct(addr, target_code_id, msg)
def _direct(self, addr: str, code_id: int, msg: Dict = None) -> Dict:
cmd = (f"msg-chaind tx wasm migrate {addr} {code_id} '{json.dumps(msg or {})}' "
f"--from admin --chain-id {self.config['chain_id']} "
f"--node {self.config['rpc']} --gas auto --gas-prices {self.config['gas_price']} -y")
return {'success': True, 'command': cmd}
def _multisig_rollback(self, addr, code_id, msg) -> Dict:
return {
'type': 'gov_proposal',
'title': f'Rollback {addr} to code {code_id}',
'messages': [{'@type': '/cosmwasm.wasm.v1.MsgMigrateContract',
'sender': 'msg1multisig...', 'contract': addr, 'code_id': str(code_id),
'msg': base64.b64encode(json.dumps(msg or {}).encode()).decode()}],
'deposit': '10000000umsg',
}
7.6 常见失败处理
| 模式 | 处理 |
|---|---|
| RPC 不可达 | 重试 3 次，指数退避 |
| Gas 不足 | 自动调整 gas_limit 重试 |
| 序列号冲突 | 等待后重试 |
| Faucet 限流 | 等待后重试，减小请求量 |
| 合约已存在 | label 追加时间戳 |
| 签失败 | 检查密钥是否正确 |
RETRY_POLICY = {'max_retries': 3, 'backoff_base': 2, 'backoff_max': 60}
def with_retry(func, *args, **kwargs):
import time, random
for attempt in range(RETRY_POLICY['max_retries'] + 1):
try: return func(*args, **kwargs)
except Exception as e:
if attempt >= RETRY_POLICY['max_retries']: raise
wait = min(RETRY_POLICY['backoff_base'] * (2 ** attempt), RETRY_POLICY['backoff_max'])
time.sleep(wait + random.uniform(0, wait * 0.1))
附录
A. 常用命令速查
# 查询链 ID
curl -s http://localhost:26657/status | jq -r '.result.node_info.network'
# 查询余额
msg-chaind query bank balances msg1... --node <rpc>
# 部署合约 (store)
msg-chaind tx wasm store contract.wasm --from deployer --chain-id msg-chain-1 --gas auto -y
# 实例化
msg-chaind tx wasm instantiate <code_id> '{"msg":{}}' --from deployer --label "x" --admin msg1... -y
# 查询合约
msg-chaind query wasm contract <addr>
# 执行方法
msg-chaind tx wasm execute <addr> '{"method":{}}' --from deployer
# 迁移合约
msg-chaind tx wasm migrate <addr> <new_code_id> '{}' --from admin
# 导出/导入密钥
msg-chaind keys export deployer --unarmored-hex
echo "mnemonic" | msg-chaind keys add deployer --recover
B. .env 模板
# .env.local
MSG_LOCAL_RPC=http://localhost:26657
MSG_LOCAL_CHAIN_ID=msg-chain-local-1
# .env.testnet
MSG_TESTNET_RPC=https://rpc-testnet.msgchain.org
MSG_TESTNET_CHAIN_ID=msg-chain-testnet-1
MSG_TESTNET_MNEMONIC=<your-mnemonic>
# .env.mainnet
MSG_MAINNET_RPC=https://rpc.msgchain.org
MSG_MAINNET_CHAIN_ID=msg-chain-1
MSG_MAINNET_HSM_KEY_ID=<hsm-key>
C. 网络端点
| 环境 | RPC | REST | gRPC | Explorer |
|---|---|---|---|---|
| Local | http://localhost:26657 |
http://localhost:1317 |
http://localhost:9090 |
无 |
| Testnet | https://rpc-testnet.msgchain.org |
https://api-testnet.msgchain.org |
https://grpc-testnet.msgchain.org |
https://explorer-testnet.msgchain.org |
| Mainnet | https://rpc.msgchain.org |
https://api.msgchain.org |
https://grpc.msgchain.org |
https://explorer.msgchain.org |
D. 参考资源
- MSG Chain 文档: https://docs.msgchain.org
- CosmWasm 文档: https://docs.cosmwasm.com
本指南旨在帮助 AI Agent 开发者进行多环境部署。所有代码仅供参考，主网部署前请务必经过安全审计。