AI Agent 快速原型与本地开发链实践指南

⚠️ No-Go Disclaimer: MSGChain 主网裁决为 No-Go。本文件所有内容反映的是开发阶段的技术设计，不代表主网未独立核验上线状态。生产部署状态请以白皮书为准：https://msgchain.org/whitepaper/

MSG Chain 上手：从零开始在测试网构建、测试和部署 AI Agent。
面向 AI Engineer / Blockchain Developer / 独立开发者

1. 概述

1.1 什么是 MSG Chain Testnet

MSG Chain 是基于 Cosmos SDK 的应用链，专为 AI Agent 经济设计。测试网与主网功能一致，但 代币零价值，可免费通过 Faucet 领取。

1.2 测试网 vs 主网

维度	测试网	主网
代币价值	零	真实 MSG
Gas 费用	免费	真实成本
部署速度	分钟级	需审计
安全要求	宽松	严格
数据	定期重置	永久

1.3 可用测试网

网络	Chain ID	RPC	Faucet
测试网	`msg-testnet-1`	`https://rpc-testnet.msgchain.org:26657`	`https://faucet.msgchain.org/claim`
本地链	`msg-local-1`	`http://localhost:26657`	无 (本地铸造)

1.4 快速路线图

0-5 min   安装依赖
5-15 min  创建钱包 + 领测试币
15-30 min 启动 Hello World Agent
30-45 min 测试 A2A 通信
45-55 min 本地链调试
55-60 min 发布到测试网

1.5 文档约定

msg1... = MSG 链 Bech32 地址
$ 开头的命令在终端执行
所有代码块可直接复制运行

2. 测试网环境准备

2.1 安装依赖

# msgd CLI
git clone https://github.com/msgchain/msg-chain.git
cd msg-chain && make install
msgd version

# Python SDK
pip install msg-sdk httpx rich asyncio pytest
python -c "import msg_sdk; print(msg_sdk.__version__)"

2.2 创建测试钱包

msgd keys add test-wallet
# 保存输出的助记词！

MSG_ADDR=$(msgd keys show test-wallet -a)
echo "Address: $MSG_ADDR"

2.3 配置测试网

msgd config chain-id msg-testnet-1
msgd config node https://rpc-testnet.msgchain.org:26657
msgd config output json
msgd config

2.4 领取测试代币

curl -X POST https://faucet.msgchain.org/claim \
  -H "Content-Type: application/json" \
  -d "{\"address\": \"$(msgd keys show test-wallet -a)\"}"

2.5 验证

msgd query bank balances $(msgd keys show test-wallet -a)
# 期望: 100000000umsg (100 MSG)

# API 连通性
curl -s https://rpc-testnet.msgchain.org:26657/status | jq .result.node_info.network
# 期望: "msg-testnet-1"

2.6 发送第一笔交易

msgd keys add test-wallet-2
MSG_ADDR_2=$(msgd keys show test-wallet-2 -a)

msgd tx bank send test-wallet $MSG_ADDR_2 1000000umsg \
  --chain-id msg-testnet-1 \
  --node https://rpc-testnet.msgchain.org:26657 \
  --fees 500umsg --gas auto --gas-adjustment 1.5 -y

2.7 环境检查脚本

echo "=== 环境检查 ==="
which msgd && msgd version || echo "❌ msgd not found"
python -c "import msg_sdk; print('✅ msg_sdk', msg_sdk.__version__)" || echo "❌ SDK not found"
msgd keys show test-wallet -a && echo "✅ Wallet exists" || echo "❌ Wallet not found"
curl -s https://rpc-testnet.msgchain.org:26657/status > /dev/null && echo "✅ RPC reachable" || echo "❌ RPC unreachable"

2.8 故障排查

# 命令找不到
export PATH=$PATH:$(go env GOPATH)/bin

# 余额为 0 → 重跑 Faucet
# Insufficient fees → --fees 500umsg
# sequence mismatch → 等待上笔交易确认再试

3. 最小可用 Agent 模板

3.1 Hello World Agent

创建 agent_hello_world.py：

#!/usr/bin/env python3
"""
最小可用 Agent —— Hello World
在 MSG Chain 测试网上注册并处理请求
"""

import asyncio, json, logging, os, time
from dataclasses import dataclass, field
from typing import Optional

try:
    from msg_sdk import AgentClient
    from msg_sdk.types import AgentStatus, Request, Response
except ImportError:
    print("请安装: pip install msg-sdk"); exit(1)

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("agent")

@dataclass
class Config:
    name: str = os.getenv("AGENT_NAME", "HelloWorldAgent")
    chain_id: str = os.getenv("CHAIN_ID", "msg-testnet-1")
    rpc: str = os.getenv("RPC_URL", "https://rpc-testnet.msgchain.org")
    mnemonic: str = os.getenv("MNEMONIC", "")
    port: int = int(os.getenv("AGENT_PORT", "8080"))
    gas_price: str = os.getenv("GAS_PRICE", "1000000000attoMSG")

class MinimalAgent:
    def __init__(self, config: Config):
        self.config = config
        self.client: Optional[AgentClient] = None
        self.agent_id: Optional[str] = None
        self.status = AgentStatus.INITIALIZING
        self._running = False
        self._req_count = 0
        self._start_time: Optional[float] = None

    async def setup(self):
        log.info(f"初始化: {self.config.name} @ {self.config.chain_id}")
        self.client = AgentClient(
            rpc=self.config.rpc, chain_id=self.config.chain_id,
            mnemonic=self.config.mnemonic, gas_price=self.config.gas_price,
        )
        chain_id = await self.client.get_chain_id()
        assert chain_id == self.config.chain_id, \
            f"链 ID 不匹配: {chain_id} != {self.config.chain_id}"
        addr = await self.client.get_address()
        bal = await self.client.get_balance(addr)
        log.info(f"地址: {addr}")
        log.info(f"余额: {bal} umsg")
        if bal == 0:
            log.warning("⚠️ 余额为 0，请先通过 Faucet 领取代币")
        self.status = AgentStatus.READY

    async def register(self, capabilities: list = None) -> str:
        caps = capabilities or ["text-generation", "echo", "ping"]
        log.info(f"注册 Agent, 能力: {caps}")
        self.status = AgentStatus.REGISTERING
        self.agent_id = await self.client.register(
            name=self.config.name,
            capabilities=caps,
            price="1000000",
            metadata={
                "version": "0.1.0",
                "description": "Hello World Agent for MSG Chain",
                "author": "test-user",
            },
        )
        log.info(f"✅ 注册成功! Agent ID: {self.agent_id}")
        self.status = AgentStatus.REGISTERED
        return self.agent_id

    async def handle_request(self, request: Request) -> Response:
        self._req_count += 1
        log.info(f"📩 [#{self._req_count}] {request.content[:60]}...")
        self.status = AgentStatus.PROCESSING
        start = time.time()
        try:
            result = await self._process(request.content)
            elapsed = time.time() - start
            return Response(
                agent_id=self.agent_id, request_id=request.id,
                content=result, status="success",
                processing_time=elapsed,
            )
        except Exception as e:
            elapsed = time.time() - start
            log.error(f"❌ 处理失败: {e}")
            return Response(
                agent_id=self.agent_id, request_id=request.id,
                content=f"Error: {str(e)}", status="error",
                processing_time=elapsed,
            )
        finally:
            self.status = AgentStatus.REGISTERED

    async def _process(self, content: str) -> str:
        """核心业务逻辑 —— 可替换为 LLM 调用"""
        await asyncio.sleep(0.1)
        if content.startswith("/echo"): return content.replace("/echo", "").strip()
        if content.startswith("/upper"): return content.replace("/upper", "").strip().upper()
        if content.startswith("/ping"): return "pong"
        if content.startswith("/reverse"): return content.replace("/reverse", "").strip()[::-1]
        if content.startswith("/len"): return str(len(content.replace("/len", "").strip()))
        return f"Hello from {self.config.name}! 消息: {content[:200]}"

    async def start(self):
        if not self.agent_id:
            raise RuntimeError("Agent 尚未注册，先调用 register()")
        self._running = True
        self._start_time = time.time()
        self.status = AgentStatus.RUNNING
        log.info(f"🚀 启动 HTTP 服务 :{self.config.port}")
        from aiohttp import web

        async def health(request):
            uptime = time.time() - self._start_time if self._start_time else 0
            return web.json_response({
                "agent_id": self.agent_id,
                "name": self.config.name,
                "status": self.status.name,
                "uptime": f"{uptime:.0f}s",
                "requests": self._req_count,
            })

        async def handler(request):
            body = await request.json()
            req = Request(
                id=body.get("id", ""),
                content=body.get("content", ""),
                sender=body.get("sender", ""),
            )
            resp = await self.handle_request(req)
            return web.json_response({
                "content": resp.content,
                "status": resp.status,
                "processing_time": resp.processing_time,
            })

        app = web.Application()
        app.router.add_get("/health", health)
        app.router.add_post("/request", handler)
        runner = web.AppRunner(app)
        await runner.setup()
        site = web.TCPSite(runner, "0.0.0.0", self.config.port)
        await site.start()
        log.info(f"🌐 Health: http://localhost:{self.config.port}/health")
        while self._running:
            await asyncio.sleep(1)

    async def stop(self):
        log.info("🛑 停止 Agent...")
        self._running = False
        self.status = AgentStatus.STOPPED

    async def unregister(self):
        if self.agent_id:
            log.info(f"注销 Agent: {self.agent_id}")
            try:
                await self.client.unregister(self.agent_id)
                log.info("✅ 注销成功")
            except Exception as e:
                log.error(f"❌ 注销失败: {e}")

    async def heartbeat(self):
        while self._running:
            try:
                await self.client.heartbeat(self.agent_id)
                log.debug("💓 Heartbeat sent")
            except Exception as e:
                log.warning(f"心跳失败: {e}")
            await asyncio.sleep(30)


async def main():
    config = Config()
    if not config.mnemonic:
        log.error("❌ 请设置 MNEMONIC 环境变量")
        return
    agent = MinimalAgent(config)
    try:
        await agent.setup()
        await agent.register()
        asyncio.create_task(agent.heartbeat())
        await agent.start()
    except KeyboardInterrupt:
        log.info("\n接收到停止信号")
    except Exception as e:
        log.error(f"致命错误: {e}")
    finally:
        await agent.stop()
        await agent.unregister()

if __name__ == "__main__":
    asyncio.run(main())

3.2 运行 Agent

# 启动（需要先设置 MNEMONIC）
export MNEMONIC="[未公开凭证]"
python agent_hello_world.py

# 另一个终端测试
curl http://localhost:8080/health | jq .

# 发送测试请求
curl -X POST http://localhost:8080/request \
  -H "Content-Type: application/json" \
  -d '{"id": "test-001", "content": "/echo Hello MSG Chain", "sender": "me"}'

# 测试更多命令
curl -X POST http://localhost:8080/request \
  -H "Content-Type: application/json" \
  -d '{"id": "t2", "content": "/ping", "sender": "me"}'

curl -X POST http://localhost:8080/request \
  -H "Content-Type: application/json" \
  -d '{"id": "t3", "content": "/reverse 测试网", "sender": "me"}'

curl -X POST http://localhost:8080/request \
  -H "Content-Type: application/json" \
  -d '{"id": "t4", "content": "/len MSG Chain is awesome", "sender": "me"}'

3.3 增强版 Agent（多能力注册表）

创建 agent_enhanced.py：

#!/usr/bin/env python3
"""
增强版 Agent —— 可扩展能力注册表 + 缓存 + A2A 通信
"""

import asyncio, hashlib, json, logging, os, time
from typing import Dict, Optional
from msg_sdk import AgentClient
from msg_sdk.types import AgentStatus, Request, Response

log = logging.getLogger("enhanced-agent")


class CapabilityRegistry:
    """可扩展能力注册表"""

    def __init__(self):
        self._handlers: Dict[str, dict] = {}

    def register(self, name: str, handler: callable, desc: str = ""):
        self._handlers[name] = {"handler": handler, "desc": desc}
        log.info(f"🛠  注册能力: /{name} — {desc}")

    def match(self, content: str) -> Optional[str]:
        for name in self._handlers:
            if content.startswith(f"/{name}"):
                return name
        return None

    def list(self) -> list:
        return [{"name": n, "desc": v["desc"]} for n, v in self._handlers.items()]


class EnhancedAgent:
    def __init__(self, config: dict):
        self.client = AgentClient(
            rpc=config.get("rpc", "https://rpc-testnet.msgchain.org"),
            chain_id=config.get("chain_id", "msg-testnet-1"),
            mnemonic=config.get("mnemonic"),
        )
        self.caps = CapabilityRegistry()
        self.agent_id = None
        self._cache = {}
        self._running = False
        self._register_defaults()

    def _register_defaults(self):
        self.caps.register("echo", lambda p: p, "回显输入")
        self.caps.register("ping", lambda p: json.dumps({"pong": True, "time": time.time()}), "健康检查")
        self.caps.register("hash", lambda p: hashlib.sha256(p.encode()).hexdigest(), "SHA256 哈希")
        self.caps.register("json", self._format_json, "JSON 格式化")
        self.caps.register("upper", lambda p: p.upper(), "转大写")
        self.caps.register("lower", lambda p: p.lower(), "转小写")
        self.caps.register("reverse", lambda p: p[::-1], "反转字符串")
        self.caps.register("translate", self._mock_translate, "模拟翻译")

    def _format_json(self, params: str) -> str:
        try:
            return json.dumps(json.loads(params), indent=2, ensure_ascii=False)
        except json.JSONDecodeError as e:
            return f"JSON 解析错误: {e}"

    async def _mock_translate(self, params: str) -> str:
        m = {"hello": "你好", "world": "世界", "thank you": "谢谢"}
        t = params.lower()
        for e, c in m.items():
            t = t.replace(e, c)
        return f"[模拟翻译] {t}"

    async def handle_request(self, request: Request) -> Response:
        cache_key = hashlib.md5(request.content.encode()).hexdigest()
        if cache_key in self._cache:
            log.info("📦 命中缓存")
            return Response(agent_id=self.agent_id, request_id=request.id,
                           content=self._cache[cache_key], status="cached")

        cap = self.caps.match(request.content)
        if not cap:
            names = ", ".join(f"/{c['name']}" for c in self.caps.list())
            return Response(agent_id=self.agent_id, request_id=request.id,
                           content=f"可用命令: {names}", status="unknown")

        handler_info = self.caps._handlers[cap]
        handler = handler_info["handler"]
        params = request.content.replace(f"/{cap}", "").strip()

        try:
            if asyncio.iscoroutinefunction(handler):
                result = await handler(params)
            else:
                result = handler(params)
            self._cache[cache_key] = result
            return Response(agent_id=self.agent_id, request_id=request.id,
                           content=result, status="success")
        except Exception as e:
            return Response(agent_id=self.agent_id, request_id=request.id,
                           content=str(e), status="error")

    async def register(self) -> str:
        names = [c["name"] for c in self.caps.list()]
        self.agent_id = await self.client.register(
            name=os.getenv("AGENT_NAME", "EnhancedAgent"),
            capabilities=names,
            price="1000000",
        )
        log.info(f"✅ 注册成功: {self.agent_id}")
        log.info(f"   能力 ({len(names)}): {', '.join(names)}")
        return self.agent_id

    async def find_peers(self, capability: str) -> list:
        registry = self.client.get_registry()
        peers = await registry.search(capability)
        log.info(f"🔍 找到 {len(peers)} 个含 '{capability}' 能力的 Agent")
        for p in peers:
            log.info(f"   - {p.name} ({str(p.id)[:16]}...)")
        return peers

    async def send_to_peer(self, peer_id: str, content: str) -> str:
        result = await self.client.send_message(target=peer_id, content=content, gas_limit=200000)
        return result

    async def start(self):
        self._running = True
        while self._running:
            await asyncio.sleep(1)

    async def stop(self):
        self._running = False


async def demo_a2a():
    """A2A 通信演示"""
    mnemonic = os.getenv("MNEMONIC")
    if not mnemonic:
        log.error("请设置 MNEMONIC")
        return

    agent_a = EnhancedAgent({"mnemonic": mnemonic, "name": "Agent-Alice"})
    agent_b = EnhancedAgent({"mnemonic": mnemonic, "name": "Agent-Bob"})

    await agent_a.client.register(name="Agent-Alice", capabilities=["echo", "ping"])
    agent_b_id = await agent_b.client.register(name="Agent-Bob", capabilities=["echo", "upper"])

    log.info(f"Agent-Bob ID: {agent_b_id}")
    result = await agent_a.client.send_message(target=agent_b_id, content="/upper hello from Alice")
    log.info(f"Bob 回复: {result}")


if __name__ == "__main__":
    asyncio.run(demo_a2a())

3.4 测试增强版 Agent

# 快速测试各能力
python -c "
import asyncio
from agent_enhanced import EnhancedAgent

async def test():
    a = EnhancedAgent({'mnemonic': '$MNEMONIC'})
    tests = [
        '/ping', '/echo hello world', '/hash msg-chain',
        '/json {\"name\":\"test\",\"value\":42}',
        '/upper hello', '/reverse 测试网',
        '/translate hello world',
        '/unknown_command',
    ]
    for cmd in tests:
        r = await a.handle_request(type('Req', (), {'id': '1', 'content': cmd, 'sender': 't'})())
        print(f'{cmd:<35} → {r.content[:70]}')

asyncio.run(test())
"

3.5 注册流程详解

Agent 注册到 MSG Chain 测试网:

1. 构建注册消息:
   - 名称 (唯一标识)
   - 能力列表 (capabilities)
   - 服务端点 (endpoint URL)
   - 定价 (price per request)
   - 元数据 (metadata)

2. 签名并广播:
   - 用 Agent 私钥签名
   - 支付 Gas 费
   - 等待 2-6 秒确认

3. 链上存储:
   - 注册到 Agent Registry 模块
   - 分配唯一 Agent ID
   - 状态 → "registered"

4. 心跳维持:
   - 每 30 秒发送心跳
   - 超过 5 分钟无心跳 → "inactive"
   - 需重新注册或恢复心跳

5. 注销:
   - 调用 unregister()
   - 释放链上存储
   - 状态 → "unregistered"

常见注册错误:
┌────────────┬────────────────────────┬──────────────────────┐
│ 错误码     │ 原因                   │ 解决                 │
├────────────┼────────────────────────┼──────────────────────┤
│ 1          │ 名称已存在             │ 更换 Agent 名称      │
│ 2          │ 余额不足               │ 通过 Faucet 领取     │
│ 3          │ Gas 不足               │ 增加 gas limit       │
│ 4          │ 签名无效               │ 检查助记词/私钥      │
│ 5          │ 链 ID 不匹配           │ msgd config chain-id │
│ 6          │ 序列号不匹配           │ 等待上笔交易确认     │
└────────────┴────────────────────────┴──────────────────────┘

3.6 请求处理扩展点

# === 接入真实 LLM ===

# 方式 1: OpenAI
import openai

async def process_with_openai(content: str) -> str:
    openai.api_key = os.getenv("OPENAI_API_KEY")
    resp = await openai.ChatCompletion.acreate(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": content}],
    )
    return resp.choices[0].message.content

# 方式 2: 本地 Ollama
import httpx

async def process_with_ollama(content: str) -> str:
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            "http://localhost:11434/api/generate",
            json={"model": "llama3", "prompt": content},
        )
        return resp.json().get("response", "")

# 方式 3: 处理管线
class Pipeline:
    def __init__(self):
        self.steps = []

    def add(self, name: str, fn: callable):
        self.steps.append((name, fn))
        return self

    async def run(self, content: str) -> str:
        result = content
        for name, fn in self.steps:
            log.info(f"🔄 {name}")
            result = await fn(result) if asyncio.iscoroutinefunction(fn) else fn(result)
        return result

# 使用管线
pipeline = Pipeline()
pipeline.add("validate", lambda x: x.strip())
pipeline.add("upper", lambda x: x.upper())
pipeline.add("simulate_llm", process_with_ollama)
result = await pipeline.run(user_input)

4. 本地开发链

4.1 为什么需要本地链

远程测试网: 接近真实环境但依赖网络、限流、3s 出块
本地开发链: 离线可用、1s 出块、零成本、完全控制

4.2 启动本地链

CHAIN_ID="msg-local-1"

# 初始化
msgd init local-devnet --chain-id $CHAIN_ID

# 创建钱包
msgd keys add dev-wallet --keyring-backend test
DEV_ADDR=$(msgd keys show dev-wallet -a --keyring-backend test)

# 配置加速
sed -i 's/timeout_commit = "3s"/timeout_commit = "1s"/' ~/.msgd/config/config.toml
sed -i 's/enable = false/enable = true/' ~/.msgd/config/app.toml
sed -i 's/enabled-unsafe-cors = false/enabled-unsafe-cors = true/' ~/.msgd/config/app.toml

# 分配代币
msgd add-genesis-account $DEV_ADDR 1000000000000umsg --keyring-backend test

# 创世交易
msgd gentx dev-wallet 100000000000umsg --chain-id $CHAIN_ID --keyring-backend test
msgd collect-gentxs

# 启动
msgd start

4.3 一键启动脚本

创建 start_local_devnet.sh：

#!/usr/bin/env bash
set -euo pipefail
CHAIN_ID="msg-local-1"
HOME_DIR="$HOME/.msgd"
KEYRING="test"

cleanup() { rm -rf $HOME_DIR/config $HOME_DIR/data; }

setup() {
    msgd init local-devnet --chain-id $CHAIN_ID --home $HOME_DIR
    echo "y" | msgd keys add dev-wallet --keyring-backend $KEYRING 2>/dev/null || true
    DEV_ADDR=$(msgd keys show dev-wallet -a --keyring-backend $KEYRING)
    msgd add-genesis-account $DEV_ADDR 1000000000000umsg --keyring-backend $KEYRING --home $HOME_DIR
    sed -i 's/timeout_commit = "3s"/timeout_commit = "1s"/' $HOME_DIR/config/config.toml
    sed -i 's/enable = false/enable = true/' $HOME_DIR/config/app.toml
    sed -i 's/enabled-unsafe-cors = false/enabled-unsafe-cors = true/' $HOME_DIR/config/app.toml
    msgd gentx dev-wallet 100000000000umsg --chain-id $CHAIN_ID --keyring-backend $KEYRING --home $HOME_DIR
    msgd collect-gentxs --home $HOME_DIR
    echo "✅ 配置完成 | Wallet: $DEV_ADDR"
}

start() {
    echo "🚀 RPC: http://localhost:26657 | API: http://localhost:1317"
    msgd start --home $HOME_DIR --rpc.laddr tcp://0.0.0.0:26657 --api.address tcp://0.0.0.0:1317
}

case "${1:-}" in clean) cleanup; setup;; *) setup; start;; esac

chmod +x start_local_devnet.sh
./start_local_devnet.sh clean

4.4 连接本地链

msgd config chain-id msg-local-1
msgd config node http://localhost:26657
msgd status | jq .node_info.network

4.5 快速迭代工作流

# 修改代码 → 重启 → 测试
CHAIN_ID="msg-local-1" RPC_URL="http://localhost:26657" python agent_hello_world.py &

curl http://localhost:8080/health | jq .
# 测试 → 改代码 → kill → 重来

# 重置本地链（完全干净状态）
pkill msgd; rm -rf ~/.msgd/data; ./start_local_devnet.sh clean

4.5 快速迭代工作流示例

# iteration_demo.py —— 在本地链上演示完整迭代流程
import asyncio, os

LOCAL_RPC = "http://localhost:26657"
TESTNET_RPC = "https://rpc-testnet.msgchain.org"

async def iterate_on_local():
    """在本地链上迭代开发"""
    from msg_sdk import AgentClient

    # 1. 连本地链
    client = AgentClient(
        rpc=LOCAL_RPC, chain_id="msg-local-1",
        mnemonic="[未公开凭证]",
    )

    # 2. 注册 Agent
    aid = await client.register(
        name=f"IterationAgent-{os.urandom(4).hex()}",
        capabilities=["text-generation", "echo"],
        price="1000",
    )
    print(f"✅ Agent 注册: {aid}")

    # 3. 测试通信
    peers = await client.get_registry().search("echo")
    print(f"🔍 发现 {len(peers)} 个 Agent")

    # 4. 测 Gas
    gas = await client.estimate_gas(
        messages=[{"type": "register_agent", "data": {"name": "gas-test", "capabilities": ["x"], "price": "1000"}}],
    )
    print(f"⛽ Gas: {gas}")

    # 5. 测转账
    tx = await client.send_tokens(
        recipient="msg1testtesttesttesttesttesttesttest",
        amount="1000umsg",
    )
    print(f"💸 交易: {tx}")

    # 6. 注销
    await client.unregister(aid)
    print("✅ Agent 已注销")


async def deploy_to_testnet():
    """准备就绪后部署到测试网"""
    from msg_sdk import AgentClient

    client = AgentClient(
        rpc=TESTNET_RPC, chain_id="msg-testnet-1",
        mnemonic=os.getenv("MNEMONIC"),
    )

    addr = await client.get_address()
    bal = await client.get_balance(addr)
    print(f"测试网地址: {addr}")
    print(f"余额: {bal} umsg")
    if int(bal) == 0:
        print("⚠️ 余额为 0，请通过 Faucet 领取")

    aid = await client.register(
        name=f"ProductionAgent-{os.urandom(4).hex()}",
        capabilities=["text-generation", "data-analysis"],
        price="5000000",
        metadata={"version": "0.2.0"},
    )
    print(f"✅ 测试网部署成功: {aid}")


if __name__ == "__main__":
    import sys
    if "--testnet" in sys.argv:
        asyncio.run(deploy_to_testnet())
    else:
        asyncio.run(iterate_on_local())

4.6 高级：本地链场景测试

# 场景 1: 测试链回滚
# 停止节点，删除最近的区块数据
pkill msgd
rm -rf ~/.msgd/data/cs.wal
msgd start --unsafe-skip-upgrades 42  # 跳过高度 42

# 场景 2: 测试竞争条件（同时发送两笔交易）
for i in 1 2; do
  msgd tx bank send dev-wallet $BOB 1000000umsg --chain-id msg-local-1 \
    --fees 500umsg --keyring-backend test -y &
done
wait

# 场景 3: 测试大交易
LARGE=$(python -c "print('x' * 100000)")
msgd tx wasm execute msg1contract... "{\"store\": \"$LARGE\"}" \
  --from dev-wallet --gas 2000000 --fees 10000umsg -y

# 场景 4: 测试高频交易（压力测试本地链）
for i in $(seq 1 20); do
  msgd tx bank send dev-wallet $ALICE 1000umsg --chain-id msg-local-1 \
    --fees 500umsg --keyring-backend test -y --broadcast-mode async
done

4.7 部署 WASM 合约到本地链

# 需要先编译 .wasm 文件
msgd tx wasm store agent_contract.wasm --from dev-wallet \
  --chain-id msg-local-1 --node http://localhost:26657 \
  --gas auto --fees 5000umsg --keyring-backend test -y
# 记录 code_id (通常是 1)

msgd tx wasm instantiate 1 '{}' --from dev-wallet \
  --label "agent-registry" --admin $DEV_ADDR \
  --chain-id msg-local-1 --node http://localhost:26657 \
  --gas auto --fees 5000umsg --keyring-backend test -y

# 查询合约状态
msgd query wasm contract-state all msg1contractaddress... --node http://localhost:26657

# 调用合约
msgd tx wasm execute msg1contract... '{"register_agent":{"name":"MyAgent","capabilities":["text"]}}' \
  --from dev-wallet --chain-id msg-local-1 --gas auto --fees 5000umsg --keyring-backend test -y

# 快速重部署（修改代码后）
msgd tx wasm store agent_contract.wasm --from dev-wallet \
  --chain-id msg-local-1 --node http://localhost:26657 \
  --gas auto --fees 5000umsg --keyring-backend test -y

4.8 本地链调试命令

msgd status                                         # 链状态和同步进度
msgd status | jq .sync_info.latest_block_height     # 当前区块高度
curl -s http://localhost:26657/unconfirmed_txs | jq .result  # 待处理交易池
curl -s http://localhost:26657/consensus_state | jq .result  # 共识状态
curl -s http://localhost:26657/health                # 节点健康检查
curl -s http://localhost:26657/net_info | jq .result  # 网络连接信息
msgd query staking validators --node http://localhost:26657  # 验证者列表

# 监视实时交易（每 2 秒刷新）
watch -n 2 'curl -s http://localhost:26657/tx_search?query="tx.height>=0" | jq ".result | {count: (.txs | length)}"'

# 查看完整配置
msgd config
cat ~/.msgd/config/config.toml | grep -v "^#\|^$"
cat ~/.msgd/config/app.toml | grep -v "^#\|^$"

4.9 重置本地链（完全恢复出厂状态）

# 完整重置脚本
pkill msgd 2>/dev/null
sleep 1
rm -rf ~/.msgd/data ~/.msgd/config
msgd init local-devnet --chain-id msg-local-1
msgd keys add dev-wallet --keyring-backend test
DEV_ADDR=$(msgd keys show dev-wallet -a --keyring-backend test)
msgd add-genesis-account $DEV_ADDR 1000000000000umsg --keyring-backend test
msgd gentx dev-wallet 100000000000umsg --chain-id msg-local-1 --keyring-backend test
msgd collect-gentxs
msgd start

5. 交互式测试脚本

5.1 完整集成测试

创建 test_agent.py：

#!/usr/bin/env python3
"""集成测试: 注册 → 发现 → 通信 → 支付 → 注销"""

import asyncio, json, logging, os, sys, time, uuid
from dataclasses import dataclass
from typing import Optional

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
log = logging.getLogger("test")

@dataclass
class TestConfig:
    chain_id: str = os.getenv("CHAIN_ID", "msg-testnet-1")
    rpc: str = os.getenv("RPC_URL", "https://rpc-testnet.msgchain.org")
    api: str = os.getenv("API_URL", "https://api-testnet.msgchain.org")
    mnemonic: str = os.getenv("MNEMONIC", "")
    faucet: str = os.getenv("FAUCET_URL", "https://faucet.msgchain.org/claim")
    name: str = f"TestAgent-{uuid.uuid4().hex[:8]}"

    def validate(self):
        if not self.mnemonic: raise ValueError("❌ 设置 MNEMONIC")

@dataclass
class Result:
    name: str; passed: bool; duration: float; error: str = None; details: dict = None
    def __str__(self):
        icon = "✅" if self.passed else "❌"
        return f"{icon} {self.name} ({self.duration:.2f}s)" + (f"\n   {self.error}" if self.error else "")

def is_valid_addr(a): return a.startswith("msg1") and len(a) >= 39
def is_positive(b): return int(b) > 0

class AgentTestSuite:
    def __init__(self, cfg: TestConfig):
        self.cfg = cfg; self.client = None; self.agent_id = None; self.addr = None
        self.results: list[Result] = []

    async def setup(self):
        log.info(f"测试套件 | {self.cfg.chain_id} | {self.cfg.name}")
        from msg_sdk import AgentClient
        self.client = AgentClient(rpc=self.cfg.rpc, chain_id=self.cfg.chain_id, mnemonic=self.cfg.mnemonic)
        self.addr = await self.client.get_address()
        log.info(f"地址: {self.addr}")

    async def test_faucet(self) -> Result:
        start = time.time()
        try:
            import httpx
            async with httpx.AsyncClient(timeout=30) as h:
                r = await h.post(self.cfg.faucet, json={"address": self.addr})
                d = r.json()
            assert r.status_code == 200 and d.get("status") == "ok"
            await asyncio.sleep(5)
            return Result("Faucet 领取代币", True, time.time()-start, details=d)
        except Exception as e:
            return Result("Faucet 领取代币", False, time.time()-start, error=str(e))

    async def test_balance(self) -> Result:
        start = time.time()
        try:
            bal = await self.client.get_balance(self.addr)
            assert is_positive(bal), f"余额 {bal}"
            return Result("余额查询", True, time.time()-start, details={"balance": f"{bal}umsg"})
        except Exception as e:
            return Result("余额查询", False, time.time()-start, error=str(e))

    async def test_register(self) -> Result:
        start = time.time()
        try:
            self.agent_id = await self.client.register(
                name=self.cfg.name, capabilities=["text-generation", "echo", "ping"], price="1000000",
                metadata={"test": True, "version": "0.1.0-test"})
            assert self.agent_id and len(self.agent_id) > 10
            return Result("Agent 注册", True, time.time()-start, details={"id": self.agent_id})
        except Exception as e:
            return Result("Agent 注册", False, time.time()-start, error=str(e))

    async def test_registry(self) -> Result:
        start = time.time()
        try:
            registry = self.client.get_registry()
            info = await registry.get_agent(self.agent_id)
            assert info and info.name == self.cfg.name
            all_a = await registry.list_agents(limit=10)
            assert len(all_a) >= 1
            return Result("Registry 查询", True, time.time()-start, details={"total": len(all_a)})
        except Exception as e:
            return Result("Registry 查询", False, time.time()-start, error=str(e))

    async def test_discovery(self) -> Result:
        start = time.time()
        try:
            peers = await self.client.get_registry().search("echo")
            found = any(p.id == self.agent_id for p in peers)
            assert found
            return Result("同伴发现", True, time.time()-start, details={"count": len(peers), "found_self": found})
        except Exception as e:
            return Result("同伴发现", False, time.time()-start, error=str(e))

    async def test_a2a(self) -> Result:
        start = time.time()
        try:
            peers = await self.client.get_registry().search("echo")
            if len(peers) < 1: raise RuntimeError("无可通信 Agent")
            result = await self.client.send_message(target=peers[0].id, content="Hello from test!")
            assert result and len(result) > 0
            return Result("A2A 通信", True, time.time()-start, details={"resp_len": len(result)})
        except Exception as e:
            return Result("A2A 通信", False, time.time()-start, error=str(e))

    async def test_heartbeat(self) -> Result:
        start = time.time()
        try:
            await self.client.heartbeat(self.agent_id)
            return Result("心跳", True, time.time()-start)
        except Exception as e:
            return Result("心跳", False, time.time()-start, error=str(e))

    async def test_gas_estimate(self) -> Result:
        start = time.time()
        try:
            g = await self.client.estimate_gas(messages=[{"type": "register_agent", "data": {"name": "gas-test", "capabilities": ["x"]}}], gas_adjustment=1.0)
            assert g > 0 and g < 10000000
            return Result("Gas 估算", True, time.time()-start, details={"gas": g})
        except Exception as e:
            return Result("Gas 估算", False, time.time()-start, error=str(e))

    async def test_unregister(self) -> Result:
        start = time.time()
        try:
            await self.client.unregister(self.agent_id)
            registry = self.client.get_registry()
            try:
                info = await registry.get_agent(self.agent_id)
                unreg = info.status == "unregistered"
            except: unreg = True
            assert unreg
            return Result("Agent 注销", True, time.time()-start)
        except Exception as e:
            return Result("Agent 注销", False, time.time()-start, error=str(e))

    async def run_all(self):
        await self.setup()
        tests = [self.test_faucet, self.test_balance, self.test_register, self.test_registry,
                 self.test_discovery, self.test_a2a, self.test_heartbeat, self.test_gas_estimate, self.test_unregister]
        for t in tests:
            r = await t(); self.results.append(r); print(str(r))
        passed = sum(1 for r in self.results if r.passed)
        total = len(self.results)
        print(f"\n📊 通过: {passed}/{total} ({passed/total*100:.1f}%)")
        if passed < total: sys.exit(1)

async def main():
    cfg = TestConfig(); cfg.validate()
    await AgentTestSuite(cfg).run_all()

if __name__ == "__main__":
    asyncio.run(main())

5.2 运行测试

MNEMONIC="abandon abandon..." python test_agent.py

5.3 冒烟测试（30 秒）

创建 smoke_test.py：

#!/usr/bin/env python3
"""30 秒冒烟测试"""
import asyncio, os, sys

async def smoke():
    from msg_sdk import AgentClient
    m = os.getenv("MNEMONIC")
    if not m: print("❌ 设置 MNEMONIC"); return False
    c = AgentClient(rpc="https://rpc-testnet.msgchain.org", chain_id="msg-testnet-1", mnemonic=m)
    checks = []
    try:
        cid = await c.get_chain_id(); ok = cid == "msg-testnet-1"
        checks.append(("连通性", ok))
    except Exception as e: checks.append(("连通性", False, str(e)))
    try:
        addr = await c.get_address(); bal = await c.get_balance(addr)
        checks.append(("余额", int(bal) > 0, f"{bal}umsg"))
    except Exception as e: checks.append(("余额", False, str(e)))
    try:
        aid = await c.register(name=f"Smoke-{os.urandom(4).hex()}", capabilities=["echo"], price="1000")
        checks.append(("注册", True, aid))
        await c.unregister(aid)
        checks.append(("注销", True))
    except Exception as e: checks.append(("注册/注销", False, str(e)))
    all_pass = True
    for ch in checks:
        s = "✅" if ch[1] else "❌"; d = f" — {ch[2]}" if len(ch)>2 else ""
        print(f"{s} {ch[0]}{d}")
        if not ch[1]: all_pass = False
    return all_pass

if __name__ == "__main__":
    sys.exit(0 if asyncio.run(smoke()) else 1)

5.4 Gas 估算工具

创建 gas_estimator.py：

#!/usr/bin/env python3
"""估算各操作 Gas 消耗"""
import asyncio, os
from msg_sdk import AgentClient

async def main():
    m = os.getenv("MNEMONIC")
    if not m: print("❌ 设置 MNEMONIC"); return
    c = AgentClient(rpc="https://rpc-testnet.msgchain.org", chain_id="msg-testnet-1", mnemonic=m)
    tests = {
        "注册(1能力)": [{"type": "register_agent", "data": {"name": "g1","capabilities":["x"],"price":"1000"}}],
        "注册(5能力)": [{"type": "register_agent", "data": {"name": "g5","capabilities":["a","b","c","d","e"],"price":"1000"}}],
        "A2A消息": [{"type": "send_message", "data": {"target":"msg1test...","content":"Hello"}}],
        "心跳": [{"type": "agent_heartbeat", "data": {"agent_id":"test"}}],
        "注销": [{"type": "unregister_agent", "data": {"agent_id":"test"}}],
        "转账": [{"type": "bank_send", "data": {"from":"msg1...","to":"msg1...","amount":"1000umsg"}}],
    }
    print(f"{'操作':<20} {'Gas':<10}")
    print("-" * 30)
    for name, msgs in tests.items():
        try:
            g = await c.estimate_gas(messages=msgs, gas_adjustment=1.0)
            print(f"{name:<20} {g:<10,}")
        except Exception as e: print(f"{name:<20} ❌ {e}")
        await asyncio.sleep(0.5)

if __name__ == "__main__":
    asyncio.run(main())

5.5 压力测试

创建 stress_test.py：

#!/usr/bin/env python3
"""A2A 并发压力测试"""
import asyncio, logging, os, time
from statistics import mean, median, stdev
from msg_sdk import AgentClient

log = logging.getLogger("stress")

async def stress(target_id: str, n: int = 50, concurrency: int = 10):
    c = AgentClient(rpc=os.getenv("RPC_URL","https://rpc-testnet.msgchain.org"), chain_id=os.getenv("CHAIN_ID","msg-testnet-1"), mnemonic=os.getenv("MNEMONIC"))
    results = []; sem = asyncio.Semaphore(concurrency)
    async def send(i):
        async with sem:
            t0 = time.time()
            try:
                r = await c.send_message(target=target_id, content=f"stress #{i}", gas_limit=300000)
                results.append({"ok": True, "t": time.time()-t0, "len": len(r)})
            except Exception as e: results.append({"ok": False, "t": time.time()-t0, "err": str(e)})
    await asyncio.gather(*[send(i) for i in range(n)])
    ok = [r for r in results if r["ok"]]
    fail = [r for r in results if not r["ok"]]
    times = [r["t"] for r in ok]
    print(f"📊 压力测试 | 总{n} 并发{concurrency}")
    print(f"   成功 {len(ok)} 失败 {len(fail)}")
    if times: print(f"   时延: min={min(times):.3f}s max={max(times):.3f}s avg={mean(times):.3f}s med={median(times):.3f}s 吞吐={len(ok)/sum(times):.1f}req/s")

if __name__ == "__main__":
    t = os.getenv("TARGET_AGENT_ID")
    if not t: print("❌ TARGET_AGENT_ID"); exit(1)
    asyncio.run(stress(t, int(os.getenv("N","50")), int(os.getenv("C","10"))))

6. 调试与日志

6.1 交易模拟

# 模拟模式，不消耗 Gas
msgd tx bank send test-wallet msg1... 1000000umsg --dry-run -y

# Python SDK
python -c "
import asyncio; from msg_sdk import AgentClient
async def sim():
    c = AgentClient(rpc='https://rpc-testnet.msgchain.org', chain_id='msg-testnet-1', mnemonic='$MNEMONIC')
    g = await c.estimate_gas(messages=[{'type':'register_agent','data':{'name':'sim-test','capabilities':['x'],'price':'1000'}}], gas_adjustment=1.3)
    print(f'估算 Gas: {g}')
asyncio.run(sim())
"

6.2 事件日志

# 查询交易事件
msgd query tx $TX_HASH --output json | jq '.logs[].events[] | select(.type == "agent_registered") | .attributes'

# WebSocket 实时监听
wscat -c wss://rpc-testnet.msgchain.org:26657/websocket
# > {"jsonrpc":"2.0","method":"subscribe","params":{"query":"tm.event='Tx'"},"id":1}

6.3 Python 事件监听

class EventMonitor:
    def __init__(self, client):
        self.client = client
        self.handlers = {}

    def on(self, event_type: str, handler: callable):
        self.handlers[event_type] = handler

    async def start(self):
        async for event in self.client.subscribe_events(query="tm.event='Tx'"):
            for log_entry in event.get("logs", []):
                for ev in log_entry.get("events", []):
                    if ev["type"] in self.handlers:
                        attrs = {a["key"]: a["value"] for a in ev.get("attributes", [])}
                        await self.handlers[ev["type"]](event["hash"], attrs)

# 使用
monitor = EventMonitor(client)
@monitor.on("agent_registered")
async def on_reg(tx_hash, attrs):
    print(f"🆕 Agent {attrs.get('name')} 注册 | tx: {tx_hash[:16]}...")
@monitor.on("agent_executed")
async def on_exec(tx_hash, attrs):
    print(f"⚡ 执行 {attrs.get('request_id')} | {attrs.get('status')}")
await monitor.start()

6.4 错误解码

# ABCI 错误码
# Code 0: 成功 | 1: 编码错 | 2: 签名失败 | 3: 序列号错
# Code 4: 余额不足 | 6: Gas 不足 | 64: Agent 未注册
# Code 65: Agent 已存在 | 66: 能力不支持

class ErrorDecoder:
    PATTERNS = {
        "already registered": ("AGENT_EXISTS", "Agent 名称已被占用"),
        "insufficient funds": ("NO_FUNDS", "余额不足，请用 Faucet"),
        "sequence mismatch": ("SEQ_MISMATCH", "等待上笔交易确认"),
        "out of gas": ("OOG", "增加 gas-limit"),
        "agent not found": ("NOT_FOUND", "Agent 不存在"),
        "signature verification failed": ("BAD_SIG", "检查助记词"),
    }

    @staticmethod
    def decode(err: str) -> str:
        for pat, (code, msg) in ErrorDecoder.PATTERNS.items():
            if pat in err.lower(): return f"[{code}] {msg}"
        return f"[未知] {err[:200]}"

6.5 RPC 健康检查

for ep in "https://rpc-testnet.msgchain.org:26657/status" \
          "https://api-testnet.msgchain.org/cosmos/base/tendermint/v1beta1/node_info"; do
  status=$(curl -s -o /dev/null -w "%{http_code}" --max-time 5 "$ep")
  echo "$([ "$status" = 200 ] && echo '✅' || echo '❌') $ep ($status)"
done

# WebSocket 测试
python3 -c "
import asyncio, json, websockets
async def t():
    async with websockets.connect('wss://rpc-testnet.msgchain.org:26657/websocket') as ws:
        await ws.send(json.dumps({'jsonrpc':'2.0','method':'subscribe','params':{'query':\"tm.event='NewBlock'\"},'id':1}))
        print('✅ WebSocket 连接成功')
asyncio.run(t())
"

6.6 日志配置

import logging, sys
from logging.handlers import RotatingFileHandler

def setup_logging(name: str = "agent", level: str = "INFO", log_file: str = None):
    logger = logging.getLogger(name)
    logger.setLevel(getattr(logging, level.upper()))
    logger.handlers.clear()
    fmt = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
    console = logging.StreamHandler(sys.stdout)
    console.setFormatter(fmt)
    logger.addHandler(console)
    if log_file:
        fh = RotatingFileHandler(log_file, maxBytes=10*1024*1024, backupCount=5)
        fh.setFormatter(logging.Formatter("%(asctime)s [%(levelname)s] %(name)s:%(lineno)d %(message)s"))
        logger.addHandler(fh)
    return logger

# JSON 格式（适合日志系统）
class JsonFormatter(logging.Formatter):
    def format(self, record):
        import json, traceback
        entry = {"timestamp": self.formatTime(record), "level": record.levelname, "logger": record.name, "message": record.getMessage()}
        if record.exc_info: entry["exception"] = "".join(traceback.format_exception(*record.exc_info))
        return json.dumps(entry, ensure_ascii=False)

7. 从测试网到主网

7.1 迁移检查清单

MIGRATION_CHECKLIST = [
    # 网络
    {"id": "NET-001", "item": "更新 RPC 到主网 endpoint", "cmd": 'export RPC_URL="https://rpc.msgchain.org:26657"', "critical": True},
    {"id": "NET-002", "item": "更新 Chain ID", "cmd": 'export CHAIN_ID="msg-chain-1"', "critical": True},
    {"id": "NET-003", "item": "更新 API 端点", "cmd": 'export API_URL="https://api.msgchain.org"', "critical": True},
    # 安全
    {"id": "SEC-001", "item": "替换测试助记词为安全密钥（硬件钱包推荐）", "cmd": "msgd keys add prod-key --ledger", "critical": True},
    {"id": "SEC-002", "item": "密钥从环境变量移到 secrets 服务", "cmd": "# AWS Secrets / Vault / 1Password CLI", "critical": True},
    {"id": "SEC-003", "item": "确保密钥不在代码仓库中", "check": "grep -r mnemonic . --include='*.py' --include='*.sh'", "critical": True},
    # 合约
    {"id": "CTR-001", "item": "验证合约地址在主网存在", "cmd": "msgd query wasm contract $ADDR --node https://rpc.msgchain.org:26657", "critical": True},
    # 资金
    {"id": "FND-001", "item": "从交易所转入真实 MSG", "critical": True},
    {"id": "FND-002", "item": "增加 Gas 上限", "cmd": 'export MAX_GAS="1000000"', "critical": False},
    # 配置
    {"id": "CFG-001", "item": "调整定价为真实价格", "cmd": 'export PRICE="5000000"', "critical": False},
    {"id": "CFG-002", "item": "启用限流", "cmd": 'export ENABLE_RATE_LIMIT=true', "critical": True},
    {"id": "CFG-003", "item": "日志级别改为 WARNING", "cmd": 'export LOG_LEVEL="WARNING"', "critical": False},
    # 监控
    {"id": "MON-001", "item": "设置健康检查告警 (PagerDuty/Slack)", "critical": True},
    {"id": "MON-002", "item": "配置 systemd/docker 自动重启", "cmd": "# 见下方 systemd 配置", "critical": True},
    # 测试
    {"id": "TST-001", "item": "测试网运行 24h 稳定测试", "critical": True},
    {"id": "TST-002", "item": "压力测试通过 (200 req, 20并发)", "critical": False},
]

def print_checklist():
    for item in MIGRATION_CHECKLIST:
        icon = "🔴" if item["critical"] else "🟡"
        print(f"  {icon} {item['id']}: {item['item']}")
        if "cmd" in item: print(f"     {item['cmd']}")
        done = input("  ✅ 已完成? (y/n): ").lower() == 'y'
        item["done"] = done

    total = len(MIGRATION_CHECKLIST)
    done = sum(1 for i in MIGRATION_CHECKLIST if i.get("done"))
    crit_total = sum(1 for i in MIGRATION_CHECKLIST if i["critical"])
    crit_done = sum(1 for i in MIGRATION_CHECKLIST if i["critical"] and i.get("done"))
    print(f"\n汇总: {done}/{total} | 关键项: {crit_done}/{crit_total}")
    if crit_done < crit_total: print("❌ 关键项未完成，建议勿上主网")
    elif done == total: print("✅ 全部就绪！")
    else: print("⚠️ 非关键项未完成，可部署")

if __name__ == "__main__":
    print_checklist()

7.2 主网配置

# .env.production（不提交到 git）
CHAIN_ID="msg-chain-1"
RPC_URL="https://rpc.msgchain.org:26657"
API_URL="https://api.msgchain.org"
MAX_GAS=1000000
GAS_PRICE="1000000000attoMSG"
GAS_ADJUSTMENT=1.5
RATE_LIMIT=50
LOG_LEVEL="WARNING"
LOG_FILE="/var/log/agent/agent.log"
AGENT_NAME="ProductionAgent-v1"
AGENT_PRICE="5000000"

7.3 Systemd 服务

sudo cat > /etc/systemd/system/msg-agent.service << 'EOF'
[Unit]
Description=MSG Chain AI Agent
After=network.target

[Service]
Type=simple
User=agent
EnvironmentFile=/etc/msg-agent/env
EnvironmentFile=/etc/msg-agent/secrets
WorkingDirectory=/opt/msg-agent
ExecStart=/usr/bin/python3 /opt/msg-agent/main.py
Restart=always
RestartSec=5
NoNewPrivileges=true
MemoryMax=2G
CPUQuota=80%

[Install]
WantedBy=multi-user.target
EOF

sudo systemctl daemon-reload && sudo systemctl enable --now msg-agent.service

7.4 Docker 部署

FROM python:3.11-slim
WORKDIR /app
RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
COPY requirements.txt . && pip install --no-cache-dir -r requirements.txt
COPY . .
RUN useradd -m -u 1000 agent && chown -R agent:agent /app
USER agent
HEALTHCHECK --interval=30s --timeout=5s --retries=3 CMD curl -f http://localhost:8080/health || exit 1
EXPOSE 8080
CMD ["python", "main.py"]

docker build -t msg-agent:latest .
docker run -d --name msg-agent --restart always --env-file .env.production -p 8080:8080 msg-agent:latest

7.5 主网 vs 测试网速查

项目	测试网	主网
Gas 价格	免费	1,000,000,000 attoMSG/gas (flat rate)
出块	~3s	~3-6s
速率限制	100 req/min	1000 req/min (需 API Key)
注册费	免费	~10 MSG (可退还存款)
密钥	可宽松	必须硬件钱包
错误容忍	高 (可重置)	低 (财务损失风险)

7.6 回滚方案

# 紧急停止
sudo systemctl stop msg-agent.service

# 注销 Agent（避免被继续调用）
msgd tx agent unregister $AGENT_ID --from prod-key --chain-id msg-chain-1 \
  --node https://rpc.msgchain.org:26657 --fees 5000umsg -y

# 回滚代码
git checkout v0.1.0
sudo systemctl restart msg-agent.service

# 验证
curl http://localhost:8080/health

# 复盘
journalctl -u msg-agent.service -n 200

7.7 部署后监控

import asyncio, os, httpx
from datetime import datetime
from msg_sdk import AgentClient

class Monitor:
    def __init__(self, cfg):
        self.client = AgentClient(rpc=cfg["rpc"], chain_id=cfg["chain_id"], mnemonic=cfg["mnemonic"])
        self.agent_id = cfg["agent_id"]
        self.webhook = cfg.get("webhook")

    async def check(self) -> dict:
        status = {"ts": datetime.utcnow().isoformat(), "healthy": True, "checks": {}}
        try:
            chain_id = await self.client.get_chain_id()
            status["checks"]["chain"] = {"ok": chain_id == "msg-chain-1", "id": chain_id}
        except Exception as e: status["healthy"] = False; status["checks"]["chain"] = {"ok": False, "e": str(e)}
        try:
            info = await self.client.get_registry().get_agent(self.agent_id)
            status["checks"]["registered"] = {"ok": info.status == "registered", "status": info.status}
            if info.status != "registered": status["healthy"] = False
        except Exception as e: status["healthy"] = False; status["checks"]["registered"] = {"ok": False, "e": str(e)}
        try:
            addr = await self.client.get_address()
            bal = await self.client.get_balance(addr)
            min_bal = int(os.getenv("MIN_BALANCE", "10000000"))
            status["checks"]["balance"] = {"ok": int(bal) > min_bal, "balance": f"{bal}umsg"}
            if int(bal) <= min_bal: status["healthy"] = False; await self.alert(f"余额不足: {bal}")
        except Exception as e: status["healthy"] = False; status["checks"]["balance"] = {"ok": False, "e": str(e)}
        try: await self.client.heartbeat(self.agent_id); status["checks"]["heartbeat"] = {"ok": True}
        except: status["healthy"] = False; status["checks"]["heartbeat"] = {"ok": False}
        return status

    async def alert(self, msg: str):
        if self.webhook:
            async with httpx.AsyncClient() as h:
                await h.post(self.webhook, json={"text": msg})

    async def loop(self, interval: int = 60):
        while True:
            s = await self.check()
            icon = "✅" if s["healthy"] else "❌"
            print(f"{icon} [{s['ts']}] {'healthy' if s['healthy'] else 'unhealthy'}")
            for name, ch in s["checks"].items():
                print(f"  {'✅' if ch.get('ok') else '❌'} {name}: {ch.get('ok','?')}")
            await asyncio.sleep(interval)

async def main():
    m = Monitor({"rpc": os.getenv("RPC_URL"), "chain_id": os.getenv("CHAIN_ID"), "mnemonic": os.getenv("MNEMONIC"), "agent_id": os.getenv("AGENT_ID"), "webhook": os.getenv("ALERT_WEBHOOK")})
    await m.loop(int(os.getenv("INTERVAL", "60")))

if __name__ == "__main__":
    asyncio.run(main())

附录

A. 常用命令速查

msgd keys list                              # 钱包列表
msgd keys add <name>                        # 创建钱包
msgd keys show <name> -a                    # 显示地址
msgd query bank balances <addr>             # 余额
msgd query tx <hash>                        # 交易详情
msgd tx bank send <from> <to> <amt>umsg     # 转账
msgd tx wasm store <file> --from <key>      # 上传合约
msgd tx wasm instantiate <code> '{}' --from # 实例化
msgd status                                 # 节点状态

B. 端口参考

端口	服务	说明
26656	P2P	节点间通信
26657	RPC	CometBFT RPC
1317	REST API	Cosmos SDK REST
9090	gRPC	Cosmos SDK gRPC
8080	Agent	AI Agent 服务

C. 推荐开发工作流

编码 → 本地链 (msg-local-1, 1s 出块)
    → 测试网 (msg-testnet-1, 24h 稳定测试)
        → 主网 (msg-chain-1, 启用监控 + 回滚方案)

D. 资源

MSG Chain 文档: https://docs.msgchain.org
Cosmos SDK: https://docs.cosmos.network
测试网浏览器: https://explorer-testnet.msgchain.org
主网浏览器: https://explorer.msgchain.org
开发者 Discord: https://discord.gg/msgchain

基于 MSG Chain 测试网实践编写。参数可能更新，以官方文档为准。

本文档内容基于 MSGChain 代码库真实状态编写，非 AI 自动生成。
主网状态: No-Go | 白皮书: https://msgchain.org/whitepaper/