AI Agent MPC 多重签名钱包接入指南

链 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. 概述
2. MPC 核心概念
3. MPC 钱包创建与管理
4. MPC 签名会话
5. MPC 钱包发送交易
6. 与 DID 集成
7. 密钥分片备份与恢复
8. 安全最佳实践
9. 完整示例项目
10. 边界与限制

附录: API 参考、错误码表、日志格式

一、概述

1.1 为什么 MPC 对 AI Agent 至关重要

AI Agent 在去中心化环境中自主执行链上操作时，面临一个根本性难题：单点故障。如果 Agent 使用单一私钥签名，任何密钥泄露都意味着完全控制权的丧失。MPC（Multi-Party Computation，多方计算）通过将私钥分片到多个参与方，解决了这一难题：

1.2 MSG Chain MPC 技术栈

MSG Chain 的 MPC 多签体系由三层构成：

Agent API 层（对外暴露）
  ├── POST /agent/v1/mpc/wallet          — 创建 MPC 钱包（受保护，Stub）
  ├── GET  /agent/v1/mpc/wallet/{id}      — 查询 MPC 钱包（公开）
  ├── POST /agent/v1/mpc/wallet/{id}/sign — 发起签名会话（受保护，Stub）
  ├── GET  /agent/v1/mpc/wallet/{id}/sign/{session_id} — 查询签名状态（公开）
  ├── POST /agent/v1/mpc/wallet/{id}/signers — 添加签名者（受保护，Stub）
  └── DELETE /agent/v1/mpc/wallet/{id}/signers/{signer_id} — 移除签名者（受保护，Stub）

合约层（链上验证）
  └── agent_mpc_v1 — MPC 合约包（部分实现）

密码学层
  └── Dilithium-5 — 后量子签名方案 + 阈值密码学扩展

1.3 当前实现状态

重要： 写路径当前在 Agent API 中标记为 X-MSG-Stub=true。agent_mpc_v1 合约状态为"部分实现"。本文档描述的内容可基于本地沙箱开发和测试，但 不能部署到生产主网（未有主网）。参见第十节。

二、MPC 核心概念

2.1 阈值密码学（Threshold Cryptography）

阈值密码学是 MPC 多签的基础。一个 t-of-n 方案意味着：

• 总共有 n 个签名者，各自持有密钥分片
• 至少 t 个签名者需要合作才能生成有效签名
• 少于 t 个签名者无法签名，也无法重组私钥

常见配置：

2.2 密钥分片（Shamir's Secret Sharing）

Shamir 的秘密共享方案是 MPC 密钥管理的数学基础：

• 将私钥 $k$ 拆分为 $n$ 个分片 $s_1, s_2, ..., s_n$
• 基于多项式 $f(x) = k + a_1x + a_2x^2 + ... + a_{t-1}x^{t-1}$
• 任意 $t$ 个分片可通过拉格朗日插值重构私钥
• 但 MPC 签名从不重构私钥 — 各分片在本地独立参与签名计算

私钥分片示意图（2-of-3）:

        原始私钥 k
      ┌────┼────┐
      ▼    ▼    ▼
  分片 s₁  s₂  s₃
    │     │     │
    │     │     │
  签名方 1  2  3
    │     │     │
    └──┬──┘     │
       │        │
   部分签名 σ₁₂     σ₃
       │        │
       └────┬───┘
            ▼
       完整签名 σ

2.3 签名轮次

MPC 签名根据协议不同可分为：

非交互式（Non-interactive）：

• 每方独立计算部分签名，无需在线协调
• 适合异步环境（如 AI Agent 各自离线签名）
• MSG Chain 的 MPC 优先支持此模式

交互式（Interactive）：

• 多方在线交换中间值，逐轮逼近最终签名
• 需要所有参与方同时在线
• 安全性更高，但延迟更大

MSG Chain 混合模式：

• 发起方提交签名请求（指定签名者顺序）
• 签名者按顺序或并行提交部分签名
• 最终由聚合器（可以是代理节点或发起方）合并为完整 Dilithium-5 签名
• 支持超时机制：session_timeout 参数

2.4 Dilithium-5 + MPC：阈值后量子签名

Dilithium-5 是 NIST 标准化的后量子签名方案。MSG Chain 将其扩展为阈值版本：

Dilithium-5 标准特性：

• 公钥大小：2,592 字节
• 签名大小：4,595 字节（~4.5KB）
• 安全强度：NIST Level 5（≥ AES-256）
• 底层困难问题：Module-LWE 和 Module-SIS

阈值 Dilithium-5 的工作方式：

┌──────────────────────────────────────────────────┐
│              标准 Dilithium-5 签名                  │
│                                                    │
│  私钥 sk ──────────────────────▶ 签名 σ             │
│                               (单个实体持有 sk)     │
└──────────────────────────────────────────────────┘

┌──────────────────────────────────────────────────┐
│           阈值 Dilithium-5 签名                    │
│                                                    │
│  分片 sk₁ ───▶ 部分签名 σ₁                         │
│  分片 sk₂ ───▶ 部分签名 σ₂        聚合              │
│  分片 sk₃ ───▶ 部分签名 σ₃ ───▶ σ = Combine(σᵢ)  │
│                                                    │
│  最终签名 σ 与标准 Dilithium-5 签名 **完全相同**    │
│  链上验证无需知道是 MPC 签名                        │
└──────────────────────────────────────────────────┘

关键优势： 聚合后的签名与标准 Dilithium-5 签名在字节层面完全一致。这意味着：

• 现有链上验证逻辑无需修改即可验证 MPC 签名
• 合约无法区分签名来自单个私钥还是 MPC 阈值签名
• 签名体积不会随签名者数量增长（而传统多签会）

2.5 安全模型

三、MPC 钱包创建与管理

3.1 创建 MPC 钱包

API 定义

端点： POST /agent/v1/mpc/wallet
访问控制： 受保护（需要 X-API-Key）
状态： Stub（X-MSG-Stub=true）
限流： 20 req/s，突发 40

请求体：

{
  "name": "agent-mpc-wallet-1",
  "threshold": 2,
  "total_signers": 3,
  "signers": [
    {
      "did": "did:msg:agent:alice",
      "pubkey": "dilithium5:9f86d081884c7d659a2feaa0c55ad015..."
    },
    {
      "did": "did:msg:agent:bob",
      "pubkey": "dilithium5:785f3ec2eb7a6c7c1f9a72f3f2b5e6d8..."
    },
    {
      "did": "did:msg:agent:carol",
      "pubkey": "dilithium5:1a2b3c4d5e6f7890abcdef1234567890..."
    }
  ],
  "chain_id": "msg-chain-1"
}

响应：

{
  "wallet_id": "mpc-wallet-uuid-abc123",
  "name": "agent-mpc-wallet-1",
  "address": "msg1mpc...derived-address",
  "threshold": 2,
  "total_signers": 3,
  "signers": [
    {"did": "did:msg:agent:alice", "pubkey": "dilithium5:...", "status": "active"},
    {"did": "did:msg:agent:bob", "pubkey": "dilithium5:...", "status": "active"},
    {"did": "did:msg:agent:carol", "pubkey": "dilithium5:...", "status": "active"}
  ],
  "chain_id": "msg-chain-1",
  "created_at": "2026-07-06T10:00:00Z",
  "status": "active"
}

Python 示例

import httpx
import uuid
from typing import Optional

API_BASE = "http://localhost:8080"
API_KEY = "your-api-key-here"

client = httpx.Client(
    base_url=API_BASE,
    headers={
        "Content-Type": "application/json",
        "X-API-Key": API_KEY,
    },
    timeout=30,
)


def create_mpc_wallet(
    name: str,
    threshold: int,
    signers: list[dict],
    chain_id: str = "msg-chain-1",
) -> dict:
    payload = {
        "name": name,
        "threshold": threshold,
        "total_signers": len(signers),
        "signers": signers,
        "chain_id": chain_id,
    }

    response = client.post("/agent/v1/mpc/wallet", json=payload)

    # Check for stub header
    is_stub = response.headers.get("X-MSG-Stub") == "true"
    if is_stub:
        print(f"[Stub] MPC wallet creation endpoint — response may not persist")

    response.raise_for_status()
    return response.json()


# Usage
alice_signer = {
    "did": "did:msg:agent:alice",
    "pubkey": "dilithium5:9f86d081884c7d659a2feaa0c55ad015...",
}
bob_signer = {
    "did": "did:msg:agent:bob",
    "pubkey": "dilithium5:785f3ec2eb7a6c7c1f9a72f3f2b5e6d8...",
}
carol_signer = {
    "did": "did:msg:agent:carol",
    "pubkey": "dilithium5:1a2b3c4d5e6f7890abcdef1234567890...",
}

try:
    wallet = create_mpc_wallet(
        name="agent-mpc-wallet-1",
        threshold=2,
        signers=[alice_signer, bob_signer, carol_signer],
    )
    print(f"Created MPC wallet: {wallet['wallet_id']}")
    print(f"Derived address: {wallet['address']}")
except httpx.HTTPStatusError as e:
    error_body = e.response.json().get("error", {})
    print(f"Error {error_body.get('code')}: {error_body.get('message')}")

TypeScript 示例

import { AgentAPIClient } from "./agent-api-client";

interface MPCWalletRequest {
  name: string;
  threshold: number;
  total_signers: number;
  signers: Array<{
    did: string;
    pubkey: string;
  }>;
  chain_id: string;
}

interface MPCWalletResponse {
  wallet_id: string;
  name: string;
  address: string;
  threshold: number;
  total_signers: number;
  signers: Array<{
    did: string;
    pubkey: string;
    status: string;
  }>;
  chain_id: string;
  created_at: string;
  status: string;
}

const API_BASE = "http://localhost:8080";
const API_KEY = "your-api-key-here";

async function createMPCWallet(
  request: MPCWalletRequest
): Promise<MPCWalletResponse> {
  const response = await fetch(`${API_BASE}/agent/v1/mpc/wallet`, {
    method: "POST",
    headers: {
      "Content-Type": "application/json",
      "X-API-Key": API_KEY,
    },
    body: JSON.stringify(request),
  });

  if (response.headers.get("X-MSG-Stub") === "true") {
    console.warn("[Stub] MPC wallet creation endpoint — may not persist");
  }

  if (!response.ok) {
    const errorBody = await response.json().catch(() => ({}));
    throw new Error(
      `MPC wallet creation failed: ${errorBody.error?.message || response.statusText}`
    );
  }

  return response.json();
}

async function main() {
  const wallet = await createMPCWallet({
    name: "agent-mpc-wallet-1",
    threshold: 2,
    total_signers: 3,
    signers: [
      { did: "did:msg:agent:alice", pubkey: "dilithium5:9f86d081..." },
      { did: "did:msg:agent:bob", pubkey: "dilithium5:785f3ec2..." },
      { did: "did:msg:agent:carol", pubkey: "dilithium5:1a2b3c4d..." },
    ],
    chain_id: "msg-chain-1",
  });

  console.log(`Created MPC wallet: ${wallet.wallet_id}`);
  console.log(`Derived address: ${wallet.address}`);
}

main().catch(console.error);

curl 示例

curl -X POST http://localhost:8080/agent/v1/mpc/wallet \
  -H "Content-Type: application/json" \
  -H "X-API-Key: your-api-key" \
  -d '{
    "name": "agent-mpc-wallet-1",
    "threshold": 2,
    "total_signers": 3,
    "signers": [
      {"did": "did:msg:agent:alice", "pubkey": "dilithium5:9f86d081..."},
      {"did": "did:msg:agent:bob", "pubkey": "dilithium5:785f3ec2..."},
      {"did": "did:msg:agent:carol", "pubkey": "dilithium5:1a2b3c4d..."}
    ],
    "chain_id": "msg-chain-1"
  }' -i

3.2 查询 MPC 钱包

API 定义

端点： GET /agent/v1/mpc/wallet/{id}
访问控制： 公开
状态： 已实现
限流： 100 req/s，突发 200

响应：

{
  "wallet_id": "mpc-wallet-uuid-abc123",
  "name": "agent-mpc-wallet-1",
  "address": "msg1mpc...",
  "threshold": 2,
  "total_signers": 3,
  "signers": [
    {"did": "did:msg:agent:alice", "pubkey": "dilithium5:...", "status": "active"},
    {"did": "did:msg:agent:bob", "pubkey": "dilithium5:...", "status": "active"},
    {"did": "did:msg:agent:carol", "pubkey": "dilithium5:...", "status": "active"}
  ],
  "balance": {"umsg": "500000000000000000"},
  "chain_id": "msg-chain-1",
  "created_at": "2026-07-06T10:00:00Z",
  "status": "active",
  "signing_sessions": [
    {
      "session_id": "sess-001",
      "status": "completed",
      "created_at": "2026-07-06T11:00:00Z"
    }
  ]
}

Python 示例

def query_mpc_wallet(wallet_id: str) -> dict:
    """Query MPC wallet details — public endpoint, no API key needed."""
    client = httpx.Client(base_url=API_BASE, timeout=30)
    response = client.get(f"/agent/v1/mpc/wallet/{wallet_id}")
    response.raise_for_status()
    return response.json()


def list_agent_mpc_wallets(agent_did: str) -> list[dict]:
    """List all MPC wallets associated with an Agent's DID."""
    client = httpx.Client(base_url=API_BASE, timeout=30)
    response = client.get(
        "/agent/v1/mpc/wallet",
        params={"agent_did": agent_did},
    )
    response.raise_for_status()
    return response.json().get("wallets", [])


# Usage
try:
    wallet = query_mpc_wallet("mpc-wallet-uuid-abc123")
    print(f"Wallet: {wallet['name']}")
    print(f"Address: {wallet['address']}")
    print(f"Balance: {wallet['balance']['umsg']} umsg")
    print(f"Threshold: {wallet['threshold']}/{wallet['total_signers']}")

    wallets = list_agent_mpc_wallets("did:msg:agent:alice")
    print(f"Agent has {len(wallets)} MPC wallets")
except httpx.HTTPStatusError as e:
    error = e.response.json().get("error", {})
    print(f"Error {error.get('code')}: {error.get('message')}")

TypeScript 示例

async function queryMPCWallet(walletId: string): Promise<any> {
  const response = await fetch(`${API_BASE}/agent/v1/mpc/wallet/${walletId}`);
  if (!response.ok) {
    const error = await response.json().catch(() => ({}));
    throw new Error(error.error?.message || "Query failed");
  }
  return response.json();
}

async function listAgentMPCWallets(agentDid: string): Promise<any[]> {
  const response = await fetch(
    `${API_BASE}/agent/v1/mpc/wallet?agent_did=${encodeURIComponent(agentDid)}`
  );
  if (!response.ok) throw new Error("Failed to list wallets");
  const data = await response.json();
  return data.wallets || [];
}

async function main() {
  const wallet = await queryMPCWallet("mpc-wallet-uuid-abc123");
  console.log(`Wallet: ${wallet.name} at ${wallet.address}`);
  console.log(`Balance: ${wallet.balance.umsg} umsg`);
}
main().catch(console.error);

curl 示例

curl -s http://localhost:8080/agent/v1/mpc/wallet/mpc-wallet-uuid-abc123 | jq '.'

3.3 添加签名者

API 定义

端点： POST /agent/v1/mpc/wallet/{id}/signers
访问控制： 受保护（需要 X-API-Key）
状态： Stub（X-MSG-Stub=true）

请求体：

{
  "signer": {
    "did": "did:msg:agent:dave",
    "pubkey": "dilithium5:abcdef1234567890..."
  }
}

响应：

{
  "wallet_id": "mpc-wallet-uuid-abc123",
  "threshold": 2,
  "total_signers": 4,
  "signers": [
    {"did": "did:msg:agent:alice", "status": "active"},
    {"did": "did:msg:agent:bob", "status": "active"},
    {"did": "did:msg:agent:carol", "status": "active"},
    {"did": "did:msg:agent:dave", "status": "pending_activation"}
  ],
  "note": "Threshold unchanged. Re-sharding may be required."
}

Python 示例

def add_mpc_signer(wallet_id: str, signer_did: str, signer_pubkey: str) -> dict:
    response = client.post(
        f"/agent/v1/mpc/wallet/{wallet_id}/signers",
        json={
            "signer": {
                "did": signer_did,
                "pubkey": signer_pubkey,
            }
        },
    )
    if response.headers.get("X-MSG-Stub") == "true":
        print("[Stub] Add signer endpoint — re-sharding simulation")
    response.raise_for_status()
    return response.json()

result = add_mpc_signer(
    "mpc-wallet-uuid-abc123",
    "did:msg:agent:dave",
    "dilithium5:abcdef1234567890...",
)
print(f"Total signers now: {result['total_signers']}")

3.4 移除签名者

API 定义

端点： DELETE /agent/v1/mpc/wallet/{id}/signers/{signer_id}
访问控制： 受保护
状态： Stub

响应：

{
  "wallet_id": "mpc-wallet-uuid-abc123",
  "threshold": 2,
  "total_signers": 2,
  "signers": [
    {"did": "did:msg:agent:alice", "status": "active"},
    {"did": "did:msg:agent:bob", "status": "active"}
  ],
  "removed_signer": "did:msg:agent:carol",
  "note": "If total_signers < threshold, wallet is locked."
}

Python 示例

def remove_mpc_signer(wallet_id: str, signer_did: str) -> dict:
    response = client.delete(
        f"/agent/v1/mpc/wallet/{wallet_id}/signers/{signer_did}"
    )
    if response.headers.get("X-MSG-Stub") == "true":
        print("[Stub] Remove signer endpoint")
    response.raise_for_status()
    return response.json()

try:
    result = remove_mpc_signer(
        "mpc-wallet-uuid-abc123", "did:msg:agent:carol",
    )
    print(f"Removed. Total signers: {result['total_signers']}")
except httpx.HTTPStatusError as e:
    error = e.response.json().get("error", {})
    if error.get("code") == 7:
        print("Invalid parameter: signer not found")
    elif error.get("code") == 5:
        print("Contract failed: re-sharding error")

3.5 错误处理矩阵

通用错误处理函数：

def handle_mpc_error(response: httpx.Response) -> dict:
    try:
        error = response.json().get("error", {})
    except Exception:
        return {"code": -1, "message": "Unknown error"}
    code = error.get("code", -1)
    message = error.get("message", "Unknown error")
    details = error.get("details", "")
    print(f"[MPC Error {code}] {message}")
    if details:
        print(f"  Details: {details}")
    if code == 3:
        raise PermissionError(f"Unauthorized: {message}")
    elif code == 17:
        raise TimeoutError(f"Rate limited: {message}")
    elif code == 7:
        raise ValueError(f"Invalid parameter: {message}")
    return error

四、MPC 签名会话

4.1 会话生命周期

MPC 签名会话遵循一个定义良好的状态机：

  CREATED
     |
     ▼
  PENDING_SIGNATURES ──────► 如果超时 ──────► TIMEOUT
     |
     ▼
  COLLECTING (部分签名到达)
     |
     ▼
  FINALIZING (聚合计算)
     |
     ├──► COMPLETED (成功)
     |
     └──► FAILED (阈值不足或验证失败)

4.2 发起签名会话

API 定义

端点： POST /agent/v1/mpc/wallet/{id}/sign
访问控制： 受保护（需要 X-API-Key）
状态： Stub（X-MSG-Stub=true）

请求体：

{
  "payload": {
    "tx_type": "cosmwasm_execute",
    "contract": "msg14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9s4hmal",
    "msg": {
      "transfer": {
        "recipient": "msg1qypqxpq9kcrn2c9afea5lq35ef37c5x7jqylz3",
        "amount": "1000000000000000000"
      }
    }
  },
  "session_timeout": 300,
  "signer_order": [
    "did:msg:agent:alice",
    "did:msg:agent:bob"
  ],
  "memo": "Payment for AI inference services",
  "chain_id": "msg-chain-1"
}

响应：

{
  "session_id": "mpc-sess-uuid-xyz789",
  "wallet_id": "mpc-wallet-uuid-abc123",
  "status": "CREATED",
  "threshold": 2,
  "signers_required": ["did:msg:agent:alice", "did:msg:agent:bob"],
  "signers_submitted": [],
  "created_at": "2026-07-06T12:00:00Z",
  "expires_at": "2026-07-06T12:05:00Z",
  "payload_hash": "sha256:abc123def456..."
}

Python 示例

import hashlib
import json


def initiate_mpc_sign(
    wallet_id: str,
    payload: dict,
    signer_order: list[str],
    session_timeout: int = 300,
    memo: str = "",
) -> dict:
    request = {
        "payload": payload,
        "session_timeout": session_timeout,
        "signer_order": signer_order,
        "memo": memo,
        "chain_id": "msg-chain-1",
    }
    response = client.post(
        f"/agent/v1/mpc/wallet/{wallet_id}/sign",
        json=request,
    )
    if response.headers.get("X-MSG-Stub") == "true":
        print("[Stub] MPC sign session endpoint")
    response.raise_for_status()
    return response.json()


payload = {
    "tx_type": "cosmwasm_execute",
    "contract": "msg14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9s4hmal",
    "msg": {
        "transfer": {
            "recipient": "msg1qypqxpq9kcrn2c9afea5lq35ef37c5x7jqylz3",
            "amount": "1000000000000000000",
        }
    },
}

session = initiate_mpc_sign(
    wallet_id="mpc-wallet-uuid-abc123",
    payload=payload,
    signer_order=["did:msg:agent:alice", "did:msg:agent:bob"],
    session_timeout=300,
    memo="Payment for AI inference services",
)
print(f"Session ID: {session['session_id']}")
print(f"Status: {session['status']}")

TypeScript 示例

interface SignSessionRequest {
  payload: { tx_type: string; contract: string; msg: Record<string, unknown> };
  session_timeout: number;
  signer_order: string[];
  memo?: string;
  chain_id: string;
}

interface SignSessionResponse {
  session_id: string;
  wallet_id: string;
  status: string;
  threshold: number;
  signers_required: string[];
  signers_submitted: string[];
  created_at: string;
  expires_at: string;
  payload_hash: string;
}

async function initiateMPCSign(
  walletId: string,
  request: SignSessionRequest
): Promise<SignSessionResponse> {
  const response = await fetch(
    `${API_BASE}/agent/v1/mpc/wallet/${walletId}/sign`,
    {
      method: "POST",
      headers: { "Content-Type": "application/json", "X-API-Key": API_KEY },
      body: JSON.stringify(request),
    }
  );
  if (!response.ok) {
    const error = await response.json().catch(() => ({}));
    throw new Error(`Failed: ${error.error?.message || response.statusText}`);
  }
  return response.json();
}

const session = await initiateMPCSign("mpc-wallet-uuid-abc123", {
  payload: {
    tx_type: "cosmwasm_execute",
    contract: "msg14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9s4hmal",
    msg: {
      transfer: {
        recipient: "msg1qypqxpq9kcrn2c9afea5lq35ef37c5x7jqylz3",
        amount: "1000000000000000000",
      },
    },
  },
  session_timeout: 300,
  signer_order: ["did:msg:agent:alice", "did:msg:agent:bob"],
  memo: "Payment for AI inference services",
  chain_id: "msg-chain-1",
});
console.log(`Session: ${session.session_id} (${session.status})`);

4.3 签名者提交部分签名

签名者 Alice 的流程:

1. 接收通知: "会话 mpc-sess-uuid-xyz789 需要你的签名"
2. 本地计算: partial_sig_alice = Sign(sk_share_alice, payload_hash)
3. 提交: POST /agent/v1/mpc/wallet/{id}/sign/{session_id}/submit
   { "signer_did": "did:msg:agent:alice", "partial_signature": "base64..." }
4. 确认: 会话状态更新为 COLLECTING

API 端点设计（提交部分签名）

端点： POST /agent/v1/mpc/wallet/{id}/sign/{session_id}/submit
访问控制： 受保护（签名者各自持 API Key）

请求体：

{
  "signer_did": "did:msg:agent:alice",
  "partial_signature": "base64-encoded-partial-sig",
  "nonce": "unique-nonce-for-replay-protection"
}

响应：

{
  "session_id": "mpc-sess-uuid-xyz789",
  "status": "COLLECTING",
  "signers_submitted": ["did:msg:agent:alice"],
  "signers_remaining": ["did:msg:agent:bob"],
  "threshold": 2
}

Python 示例（签名者提交）

import secrets

def submit_partial_signature(
    wallet_id: str,
    session_id: str,
    signer_did: str,
    partial_sig: str,
    signer_api_key: str,
) -> dict:
    signer_client = httpx.Client(
        base_url=API_BASE,
        headers={
            "Content-Type": "application/json",
            "X-API-Key": signer_api_key,
        },
        timeout=30,
    )
    response = signer_client.post(
        f"/agent/v1/mpc/wallet/{wallet_id}/sign/{session_id}/submit",
        json={
            "signer_did": signer_did,
            "partial_signature": partial_sig,
            "nonce": secrets.token_hex(16),
        },
    )
    response.raise_for_status()
    return response.json()


def generate_partial_sig(
    private_key_share: bytes,
    payload_to_sign: dict,
) -> str:
    """Generate a partial Dilithium-5 signature using the signer's key share.

    NOTE: This is a simplified representation. The actual threshold Dilithium-5
    partial signature generation uses a specialized cryptographic protocol.
    """
    import base64
    payload_bytes = json.dumps(payload_to_sign, sort_keys=True).encode("utf-8")
    payload_hash = hashlib.sha256(payload_bytes).digest()
    # In production: partial_sig = Dilithium5ThresholdSign(private_key_share, payload_hash, signer_index)
    data = private_key_share[:32] + payload_hash
    return base64.b64encode(data).decode("ascii")

# Alice submits her partial signature
alice_partial_sig = generate_partial_sig(
    b"alice-private-key-share-32-bytes...", payload
)
result = submit_partial_signature(
    wallet_id="mpc-wallet-uuid-abc123",
    session_id=session["session_id"],
    signer_did="did:msg:agent:alice",
    partial_sig=alice_partial_sig,
    signer_api_key="alice-api-key",
)
print(f"Session status: {result['status']}")
print(f"Remaining signers: {result['signers_remaining']}")

TypeScript 示例（签名者提交）

async function submitPartialSignature(
  walletId: string,
  sessionId: string,
  signerDid: string,
  partialSignature: string,
  signerApiKey: string
): Promise<any> {
  const response = await fetch(
    `${API_BASE}/agent/v1/mpc/wallet/${walletId}/sign/${sessionId}/submit`,
    {
      method: "POST",
      headers: { "Content-Type": "application/json", "X-API-Key": signerApiKey },
      body: JSON.stringify({
        signer_did: signerDid,
        partial_signature: partialSignature,
        nonce: crypto.randomUUID(),
      }),
    }
  );
  if (!response.ok) throw new Error("Failed to submit partial signature");
  return response.json();
}

const bobResult = await submitPartialSignature(
  "mpc-wallet-uuid-abc123",
  session.session_id,
  "did:msg:agent:bob",
  bobPartialSig,
  "bob-api-key"
);
console.log(`Session status: ${bobResult.status}`);

4.4 查询签名会话状态

API 定义

端点： GET /agent/v1/mpc/wallet/{id}/sign/{session_id}
访问控制： 公开
状态： 已实现

响应（进行中）：

{
  "session_id": "mpc-sess-uuid-xyz789",
  "wallet_id": "mpc-wallet-uuid-abc123",
  "status": "COLLECTING",
  "threshold": 2,
  "signers_required": ["did:msg:agent:alice", "did:msg:agent:bob"],
  "signers_submitted": ["did:msg:agent:alice"],
  "signers_remaining": ["did:msg:agent:bob"],
  "payload_hash": "sha256:abc123def456...",
  "created_at": "2026-07-06T12:00:00Z",
  "expires_at": "2026-07-06T12:05:00Z",
  "error": null
}

响应（COMPLETED）：

{
  "session_id": "mpc-sess-uuid-xyz789",
  "status": "COMPLETED",
  "full_signature": "dilithium5:base64-encoded-full-sig...",
  "public_key": "dilithium5:base64-encoded-public-key...",
  "signature_size_bytes": 4595,
  "algorithm": "Dilithium5",
  "completed_at": "2026-07-06T12:02:30Z"
}

响应（FAILED）：

{
  "session_id": "mpc-sess-uuid-xyz789",
  "status": "FAILED",
  "error": { "code": 15, "message": "Partial signature verification failed", "details": "..." },
  "created_at": "2026-07-06T12:00:00Z",
  "failed_at": "2026-07-06T12:03:00Z"
}

Python 轮询示例

import time

def poll_mpc_session(
    wallet_id: str,
    session_id: str,
    poll_interval: float = 2.0,
    timeout: float = 300.0,
) -> dict:
    start = time.time()
    while True:
        elapsed = time.time() - start
        if elapsed > timeout:
            raise TimeoutError(f"MPC session {session_id} timed out")
        response = client.get(f"/agent/v1/mpc/wallet/{wallet_id}/sign/{session_id}")
        response.raise_for_status()
        state = response.json()
        status = state["status"]
        if status in ("COMPLETED", "FAILED", "TIMEOUT"):
            print(f"Session {session_id} final status: {status}")
            return state
        remaining = state.get("signers_remaining", [])
        if remaining:
            print(f"  Waiting for signers: {remaining}")
        time.sleep(poll_interval)

final_state = poll_mpc_session(
    wallet_id="mpc-wallet-uuid-abc123",
    session_id=session["session_id"],
)
if final_state["status"] == "COMPLETED":
    print(f"Signature obtained ({final_state['signature_size_bytes']} bytes)")
elif final_state["status"] == "FAILED":
    print(f"Signing failed: {final_state['error']['message']}")

TypeScript 轮询示例

async function pollMPCSession(
  walletId: string, sessionId: string,
  pollInterval: number = 2000, timeout: number = 300000
): Promise<any> {
  const start = Date.now();
  while (true) {
    if (Date.now() - start > timeout)
      throw new Error(`Session ${sessionId} timed out`);
    const response = await fetch(`${API_BASE}/agent/v1/mpc/wallet/${walletId}/sign/${sessionId}`);
    if (!response.ok) throw new Error(`Query failed: ${response.statusText}`);
    const state = await response.json();
    if (["COMPLETED", "FAILED", "TIMEOUT"].includes(state.status)) return state;
    if (state.signers_remaining?.length > 0)
      console.log(`Waiting for: ${state.signers_remaining.join(", ")}`);
    await new Promise(r => setTimeout(r, pollInterval));
  }
}

const finalState = await pollMPCSession("mpc-wallet-uuid-abc123", session.session_id);
if (finalState.status === "COMPLETED") console.log(`Full signature: ${finalState.full_signature}`);

4.5 WebSocket 事件订阅

WebSocket 事件格式

{
  "type": "mpc_session_update",
  "data": {
    "wallet_id": "mpc-wallet-uuid-abc123",
    "session_id": "mpc-sess-uuid-xyz789",
    "previous_status": "PENDING_SIGNATURES",
    "current_status": "COLLECTING",
    "signer_submitted": "did:msg:agent:alice",
    "signers_remaining": ["did:msg:agent:bob"]
  },
  "timestamp": 1720274400
}

Python WebSocket 监听器

import asyncio
import json
import httpx

async def listen_mpc_session_ws(
    session_id: str,
    callback_status_change,
    ws_url: str = "ws://localhost:8080/agent/v1/events/subscribe",
):
    async with httpx.AsyncClient() as http_client:
        async with http_client.stream(
            "GET",
            ws_url.replace("ws://", "http://").replace("wss://", "https://"),
            params={"session_id": session_id, "event_types": "mpc_session"},
        ) as response:
            async for line in response.aiter_lines():
                if not line:
                    continue
                try:
                    event = json.loads(line)
                    if event.get("type") != "mpc_session_update":
                        continue
                    data = event.get("data", {})
                    if data.get("session_id") != session_id:
                        continue
                    await callback_status_change(
                        session_id=data["session_id"],
                        old_status=data.get("previous_status"),
                        new_status=data.get("current_status"),
                        data=data,
                    )
                except json.JSONDecodeError:
                    continue

async def on_status_change(session_id, old_status, new_status, data):
    if new_status == "COMPLETED":
        print(f"Session {session_id} completed!")
    elif new_status == "FAILED":
        print(f"Session {session_id} failed!")
    elif new_status == "COLLECTING":
        print(f"Signer submitted: {data.get('signer_submitted')}")

asyncio.run(listen_mpc_session_ws(
    session_id="mpc-sess-uuid-xyz789",
    callback_status_change=on_status_change,
))

TypeScript WebSocket 监听器

interface MPCSessionEvent {
  type: "mpc_session_update";
  data: {
    wallet_id: string; session_id: string;
    previous_status: string; current_status: string;
    signer_submitted?: string; signers_remaining?: string[];
  };
  timestamp: number;
}

class MPCSessionMonitor {
  private ws: WebSocket | null = null;
  private sessionId: string;
  private onComplete: (signature: string) => void;
  private onError: (error: string) => void;

  constructor(sessionId: string, callbacks: { onComplete: (sig: string) => void; onError: (err: string) => void }) {
    this.sessionId = sessionId;
    this.onComplete = callbacks.onComplete;
    this.onError = callbacks.onError;
  }

  connect(baseUrl: string) {
    const wsUrl = baseUrl.replace(/^http/, "ws");
    this.ws = new WebSocket(`${wsUrl}/agent/v1/events/subscribe`);
    this.ws.onopen = () => {
      this.ws!.send(JSON.stringify({ event_types: ["mpc_session_update"], filter: { session_id: this.sessionId } }));
    };
    this.ws.onmessage = (event) => {
      const msg: MPCSessionEvent = JSON.parse(event.data);
      if (msg.data.session_id !== this.sessionId) return;
      if (msg.data.current_status === "COMPLETED") this.fetchFinalSignature();
      else if (msg.data.current_status === "FAILED") this.onError("Session failed");
    };
  }

  private async fetchFinalSignature() {
    const response = await fetch(`${API_BASE}/agent/v1/mpc/session/${this.sessionId}/signature`);
    const data = await response.json();
    this.onComplete(data.full_signature);
  }

  disconnect() { this.ws?.close(); this.ws = null; }
}

const monitor = new MPCSessionMonitor("mpc-sess-uuid-xyz789", {
  onComplete: (sig) => console.log(`Got signature`),
  onError: (err) => console.error(err),
});
monitor.connect("http://localhost:8080");

4.6 完整签名流程（Python 多参与者协调）

import asyncio
import time
import hashlib
import base64
import uuid
from dataclasses import dataclass
from enum import Enum
from typing import Optional


class SessionStatus(Enum):
    CREATED = "CREATED"
    PENDING_SIGNATURES = "PENDING_SIGNATURES"
    COLLECTING = "COLLECTING"
    FINALIZING = "FINALIZING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"
    TIMEOUT = "TIMEOUT"


@dataclass
class SignerConfig:
    did: str
    api_key: str
    private_key_share: bytes


class MPCCoordinator:
    """Coordinates an MPC signing session across multiple signers."""

    def __init__(
        self,
        wallet_id: str,
        threshold: int,
        signers: list[SignerConfig],
        api_base: str = "http://localhost:8080",
    ):
        self.wallet_id = wallet_id
        self.threshold = threshold
        self.signers = {s.did: s for s in signers}
        self.api_base = api_base
        self._client = httpx.Client(base_url=api_base, timeout=30)

    def _signer_client(self, signer: SignerConfig) -> httpx.Client:
        return httpx.Client(
            base_url=self.api_base,
            headers={"Content-Type": "application/json", "X-API-Key": signer.api_key},
            timeout=30,
        )

    def initiate_session(self, payload: dict, signer_order: list[str], session_timeout: int = 300) -> dict:
        first_signer = self.signers[signer_order[0]]
        client = self._signer_client(first_signer)
        response = client.post(
            f"/agent/v1/mpc/wallet/{self.wallet_id}/sign",
            json={"payload": payload, "session_timeout": session_timeout,
                  "signer_order": signer_order, "chain_id": "msg-chain-1"},
        )
        response.raise_for_status()
        return response.json()

    def signer_submit(self, signer: SignerConfig, session_id: str, payload_hash: str) -> dict:
        partial_sig = base64.b64encode(
            signer.private_key_share[:32] + payload_hash.encode()
        ).decode("ascii")
        client = self._signer_client(signer)
        response = client.post(
            f"/agent/v1/mpc/wallet/{self.wallet_id}/sign/{session_id}/submit",
            json={"signer_did": signer.did, "partial_signature": partial_sig, "nonce": uuid.uuid4().hex},
        )
        response.raise_for_status()
        return response.json()

    def query_session(self, session_id: str) -> dict:
        response = self._client.get(f"/agent/v1/mpc/wallet/{self.wallet_id}/sign/{session_id}")
        response.raise_for_status()
        return response.json()

    def execute_full_sign_flow(
        self, payload: dict, signer_order: list[str], session_timeout: int = 300,
    ) -> Optional[dict]:
        print("[1/4] Initiating signing session...")
        session = self.initiate_session(payload, signer_order, session_timeout)
        session_id = session["session_id"]
        payload_hash = session["payload_hash"]
        print(f"  Session: {session_id}")

        print("[2/4] Collecting partial signatures...")
        for did in signer_order:
            signer = self.signers.get(did)
            if not signer:
                continue
            try:
                result = self.signer_submit(signer, session_id, payload_hash)
                print(f"  {did}: {result['status']}")
                if result["status"] == SessionStatus.COMPLETED.value:
                    break
            except Exception as e:
                print(f"  {did} failed: {e}")

        print("[3/4] Finalizing...")
        start = time.time()
        while True:
            if time.time() - start > session_timeout:
                print("  Timeout")
                return None
            state = self.query_session(session_id)
            status = state["status"]
            if status == SessionStatus.COMPLETED.value:
                print("[4/4] Complete!")
                return state
            if status in (SessionStatus.FAILED.value, SessionStatus.TIMEOUT.value):
                print(f"  Failed: {state.get('error', {}).get('message')}")
                return None
            time.sleep(2)


async def main():
    alice = SignerConfig(did="did:msg:agent:alice", api_key="alice-api-key", private_key_share=b"alice-share-32-bytes...")
    bob = SignerConfig(did="did:msg:agent:bob", api_key="bob-api-key", private_key_share=b"bob-share-32-bytes.....")
    carol = SignerConfig(did="did:msg:agent:carol", api_key="carol-api-key", private_key_share=b"carol-share-32-bytes.")

    coord = MPCCoordinator(wallet_id="mpc-wallet-uuid-abc123", threshold=2, signers=[alice, bob, carol])
    result = coord.execute_full_sign_flow(
        payload={"tx_type": "cosmwasm_execute", "contract": "msg14hj2tavq8fpesdwxxcu44rty3hh90vhujrvcmstl4zr3txmfvw9s4hmal", "msg": {"action": "ping"}},
        signer_order=["did:msg:agent:alice", "did:msg:agent:bob"],
    )
    if result:
        print(f"Signature: {result['full_signature'][:64]}...")

asyncio.run(main())

4.7 超时与错误处理

class MPCSessionError(Exception): pass
class MPCSessionTimeoutError(MPCSessionError): pass
class MPCSessionFailedError(MPCSessionError):
    def __init__(self, code: int, message: str, details: str = ""):
        super().__init__(f"[{code}] {message}")
        self.code = code; self.details = details

def handle_session_timeout(session_id: str, wallet_id: str):
    try:
        state = client.get(f"/agent/v1/mpc/wallet/{wallet_id}/sign/{session_id}").json()
        print(f"Final state: {state.get('status')}")
    except Exception:
        pass
    retry_delays = [5, 15, 45, 120, 300]
    for attempt, delay in enumerate(retry_delays):
        print(f"Retry {attempt+1} in {delay}s...")
        time.sleep(delay)
        try:
            return create_new_session()
        except Exception as e:
            print(f"Retry failed: {e}")
    raise MPCSessionTimeoutError(f"All retries exhausted")

def handle_invalid_partial_sig(session_id: str, failed_signer_did: str, error: dict):
    code = error.get("code", -1)
    if code == 15:
        log_audit_event("mpc_invalid_partial_sig", session_id=session_id, signer=failed_signer_did, error=error.get("message"))
        raise MPCSessionFailedError(15, f"Signer {failed_signer_did} invalid signature")
    else:
        return True  # retryable

def log_audit_event(event_type: str, **kwargs):
    import json
    entry = {"event_type": event_type, "timestamp": time.time(), **kwargs}
    print(f"[AUDIT] {json.dumps(entry)}")

五、MPC 钱包发送交易

5.1 完整流程概述

Phase 1: 构建交易 → Phase 2: MPC 签名 → Phase 3: 组装广播 → Phase 4: 确认

5.2 构建 Cosmos SDK 交易

from dataclasses import dataclass

@dataclass
class Coin:
    denom: str
    amount: str

@dataclass
class Fee:
    amount: list[Coin]
    gas_limit: int
    payer: str = ""
    granter: str = ""


def build_unsigned_tx_body(messages: list[dict], memo: str = "", timeout_height: int = 0) -> dict:
    return {"messages": messages, "memo": memo, "timeout_height": str(timeout_height)}


def build_auth_info(signer_address: str, sequence: int, fee: Fee, mode: str = "SIGN_MODE_DIRECT") -> dict:
    return {
        "signer_infos": [{"public_key": None, "mode_info": {"single": {"mode": mode}}, "sequence": str(sequence)}],
        "fee": {"amount": [{"denom": c.denom, "amount": c.amount} for c in fee.amount],
                "gas_limit": str(fee.gas_limit), "payer": fee.payer, "granter": fee.granter},
    }


def compute_sign_doc(tx_body: dict, auth_info: dict, chain_id: str = "msg-chain-1", account_number: int = 0) -> bytes:
    import hashlib, json
    sign_doc = {"body": tx_body, "auth_info": auth_info, "chain_id": chain_id, "account_number": str(account_number)}
    return hashlib.sha256(json.dumps(sign_doc, sort_keys=True).encode()).digest()

5.3 Gas 估算与 MPC 开销

def estimate_mpc_gas(message_count: int = 1, data_size_bytes: int = 256) -> int:
    base_gas = 100_000
    per_message_gas = message_count * 40_000
    per_data_gas = data_size_bytes * 10
    sig_verification_gas = 50_000
    return base_gas + per_message_gas + per_data_gas + sig_verification_gas

gas_limit = estimate_mpc_gas(1, 256)
fee_attoMSG = int(gas_limit * 1000000000)
print(f"Gas: {gas_limit}, Fee: {fee_attoMSG} attoMSG ({fee_attoMSG / 10**18} MSG)")

5.4 完整端到端示例

import asyncio
import base64
import time
from typing import Optional


class MPCTransactionExecutor:
    def __init__(self, api_base: str, rpc_endpoint: str, chain_id: str = "msg-chain-1"):
        self.api_base = api_base
        self.rpc_endpoint = rpc_endpoint
        self.chain_id = chain_id

    async def build_and_sign(
        self, wallet_id: str, messages: list[dict], signer_order: list[str],
        signer_configs: dict[str, SignerConfig], fee_gas_limit: int = 350_000, memo: str = "",
    ) -> Optional[bytes]:
        tx_body = build_unsigned_tx_body(messages, memo)
        auth_info = build_auth_info("", 0, Fee(
            amount=[Coin(denom="umsg", amount=str(int(fee_gas_limit * 1000000000)))],
            gas_limit=fee_gas_limit,
        ))
        sign_doc_bytes = compute_sign_doc(tx_body, auth_info, self.chain_id)

        coord = MPCCoordinator(wallet_id=wallet_id, threshold=len(signer_order),
                                signers=list(signer_configs.values()), api_base=self.api_base)
        session = coord.initiate_session(
            payload={"sign_doc_hash": hashlib.sha256(sign_doc_bytes).hexdigest()},
            signer_order=signer_order, session_timeout=300,
        )
        for did in signer_order:
            config = signer_configs.get(did)
            if not config: continue
            coord.signer_submit(config, session["session_id"], session["payload_hash"])
        final = coord.query_session(session["session_id"])
        start = time.time()
        while final["status"] not in ("COMPLETED", "FAILED", "TIMEOUT"):
            if time.time() - start > 300: raise TimeoutError("MPC signing timed out")
            await asyncio.sleep(2)
            final = coord.query_session(session["session_id"])
        if final["status"] != "COMPLETED":
            raise RuntimeError(f"MPC signing failed: {final.get('error', {}).get('message')}")
        return final["full_signature"]

    async def broadcast_tx(self, signed_tx_bytes: bytes, mode: str = "BROADCAST_MODE_SYNC") -> dict:
        tx_b64 = base64.b64encode(signed_tx_bytes).decode("ascii")
        response = httpx.Client(base_url=self.rpc_endpoint, timeout=60).post("/", json={
            "jsonrpc": "2.0", "id": 1, "method": "broadcast_tx_sync", "params": {"tx": tx_b64},
        })
        response.raise_for_status()
        result = response.json()
        tx_hash = result.get("result", {}).get("tx_hash", "")
        return {"tx_hash": tx_hash, "code": result.get("result", {}).get("code", 0)}

    async def wait_for_tx(self, tx_hash: str, timeout: int = 60) -> dict:
        start = time.time()
        while time.time() - start < timeout:
            r = httpx.Client(base_url=self.rpc_endpoint, timeout=30).post("/", json={
                "jsonrpc": "2.0", "id": 1, "method": "tx", "params": {"hash": f"0x{tx_hash}"},
            })
            if r.status_code == 200:
                tx = r.json().get("result", {})
                if tx.get("height"):
                    return {"height": int(tx["height"]), "gas_used": tx.get("gas_used", 0), "tx_hash": tx_hash}
            await asyncio.sleep(2)
        raise TimeoutError(f"Tx {tx_hash} not confirmed in {timeout}s")

    async def execute(self, wallet_id: str, messages: list[dict], signer_order: list[str],
                      signer_configs: dict[str, SignerConfig], memo: str = "") -> dict:
        signed = await self.build_and_sign(wallet_id, messages, signer_order, signer_configs, memo=memo)
        if not signed: raise RuntimeError("No signature obtained")
        broadcast = await self.broadcast_tx(signed)
        receipt = await self.wait_for_tx(broadcast["tx_hash"])
        print(f"Confirmed at height {receipt['height']}")
        return receipt


async def run_e2e():
    alice = SignerConfig("did:msg:agent:alice", "alice-api-key", b"alice-share-32bytes")
    bob = SignerConfig("did:msg:agent:bob", "bob-api-key", b"bob-share-32bytes")
    signers = {"did:msg:agent:alice": alice, "did:msg:agent:bob": bob}

    executor = MPCTransactionExecutor(api_base="http://localhost:8080", rpc_endpoint="http://localhost:26657")
    receipt = await executor.execute(
        wallet_id="mpc-wallet-uuid-abc123",
        messages=[{"@type": "/cosmos.bank.v1beta1.MsgSend", "from_address": "msg1mpc...",
                    "to_address": "msg1qypqxpq9kcrn2c9afea5lq35ef37c5x7jqylz3",
                    "amount": [{"denom": "umsg", "amount": "1000000000000000000"}]}],
        signer_order=["did:msg:agent:alice", "did:msg:agent:bob"],
        signer_configs=signers,
        memo="MPC-signed transfer",
    )
    print(f"Done at height {receipt['height']}")

asyncio.run(run_e2e())

5.5 错误恢复与重试

class RetryableMPCError(Exception): pass
class FatalMPCError(Exception): pass

async def execute_with_retry(executor, wallet_id, messages, signer_order, signer_configs, max_retries=3):
    import random
    for attempt in range(1, max_retries + 1):
        try:
            return await executor.execute(wallet_id, messages, signer_order, signer_configs)
        except RetryableMPCError as e:
            if attempt == max_retries: raise
            delay = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(delay)
        except FatalMPCError:
            raise
    raise RuntimeError("All retries exhausted")

def handle_broadcast_error(error_body: dict):
    code = error_body.get("code", -1)
    if code == 4: raise FatalMPCError("Insufficient funds")
    elif code == 5: raise RetryableMPCError(f"Contract failed: {error_body}")
    elif code == 16: raise RetryableMPCError("Nonce conflict")
    elif code == 17: raise RetryableMPCError("Rate limited")
    elif code == 15: raise FatalMPCError("Invalid MPC signature")
    else: raise RetryableMPCError(f"Error {code}")

5.6 多签名者协调方式

方式 1：HTTP 轮询 — 发起方轮询 GET /sign/{session_id}

方式 2：WebSocket 监听 — 连接 /events/subscribe 监听 mpc_session_update

方式 3：回调 URL

def initiate_with_callback(wallet_id: str, payload: dict, signer_order: list[str], callback_url: str) -> dict:
    response = client.post(
        f"/agent/v1/mpc/wallet/{wallet_id}/sign",
        json={"payload": payload, "session_timeout": 300, "signer_order": signer_order,
              "chain_id": "msg-chain-1", "callback_url": callback_url},
    )
    response.raise_for_status()
    return response.json()

# FastAPI callback handler
from fastapi import FastAPI, Request
app = FastAPI()

@app.post("/mpc-callback")
async def mpc_callback(request: Request):
    data = await request.json()
    if data.get("status") == "COMPLETED":
        print(f"Session {data['session_id']} completed!")
        # Broadcast transaction
    return {"ok": True}

六、与 DID 集成

6.1 DID-MPC 绑定架构

DID 文档 (aidid_did_registry_v1)
  |
  ├── verificationMethod ──► MPC Wallet 聚合公钥
  |     type: "Dilithium5ThresholdVerificationKey2026"
  |     publicKeyMultibase: (MPC aggregated public key)
  |
  ├── authentication ──► 引用 MPC verificationMethod
  |
  └── service ──► MPC Wallet API Endpoint
        type: "MPCWalletService"
        serviceEndpoint: "https://api.msgchain.org/agent/v1/mpc/wallet/{id}"

6.2 链接 MPC 钱包到 Agent DID

DID_REGISTRY_ADDRESS = "msg1...aididRegistryAddress"

def link_mpc_wallet_to_did(signing_client: httpx.Client, agent_did: str, mpc_wallet_info: dict, dilithium_pubkey: str) -> dict:
    resolve_response = signing_client.get(
        f"/agent/v1/query/contract/{DID_REGISTRY_ADDRESS}",
        params={"query": json.dumps({"resolve_did": {"did": agent_did}})},
    )
    doc = resolve_response.json().get("document", {})

    vm_id = f"{agent_did}#mpc-wallet-{mpc_wallet_info['wallet_id']}"
    doc.setdefault("verification_method", []).append({
        "id": vm_id, "controller": agent_did,
        "type_": "Dilithium5ThresholdVerificationKey2026",
        "public_key_multibase": dilithium_pubkey,
    })
    doc.setdefault("authentication", []).append(vm_id)
    doc.setdefault("service", []).append({
        "id": f"{agent_did}#mpc-service",
        "type_": "MPCWalletService",
        "service_endpoint": f"https://api.msgchain.org/agent/v1/mpc/wallet/{mpc_wallet_info['wallet_id']}",
    })

    response = signing_client.post(
        f"/agent/v1/wallet/transfer",
        json={"contract": DID_REGISTRY_ADDRESS, "msg": {"update_did": {"did": agent_did, "document": doc, "signature": "..."}}},
    )
    response.raise_for_status()
    return response.json()

# Usage
mpc_wallet = {"wallet_id": "mpc-wallet-uuid-abc123", "address": "msg1mpc...", "threshold": 2, "total_signers": 3}
link_mpc_wallet_to_did(auth_client, "did:msg:agent:alice", mpc_wallet, "dilithium5:base64-aggregated-pubkey...")

TypeScript 版本

const DID_CONTRACT = "msg1...aididRegistryAddress";

async function linkMPCWalletToDID(
  signingClient: SigningCosmWasmClient, sender: string,
  agentDid: string, mpcWalletId: string, mpcAddress: string, mpcPubkey: string
) {
  const doc = await signingClient.queryContractSmart(DID_CONTRACT, { resolve_did: { did: agentDid } });
  const document = doc.document;
  const vmId = `${agentDid}#mpc-wallet-${mpcWalletId}`;

  document.verification_method.push({
    id: vmId, controller: agentDid,
    type_: "Dilithium5ThresholdVerificationKey2026",
    public_key_multibase: mpcPubkey,
  });
  document.authentication.push(vmId);
  if (!document.service) document.service = [];
  document.service.push({
    id: `${agentDid}#mpc-service`, type_: "MPCWalletService",
    service_endpoint: `https://api.msgchain.org/agent/v1/mpc/wallet/${mpcWalletId}`,
  });

  return signingClient.execute(sender, DID_CONTRACT, { update_did: { did: agentDid, document, signature: "..." } }, "auto");
}

6.3 使用 MPC 签名进行 DID 认证

def prove_mpc_wallet_control(agent_did: str, wallet_id: str, challenge: str, mpc_signature: str) -> bool:
    response = client.get(
        f"/agent/v1/query/contract/{DID_REGISTRY_ADDRESS}",
        params={"query": json.dumps({"resolve_did": {"did": agent_did}})},
    )
    doc = response.json().get("document", {})
    mpc_vm = next((vm for vm in doc.get("verification_method", []) if "mpc-wallet" in vm.get("id", "")), None)
    if not mpc_vm:
        return False
    pubkey = mpc_vm.get("public_key_multibase", "")
    verify_response = client.post("/agent/v1/query/verify", json={
        "algorithm": "Dilithium5", "public_key": pubkey, "signature": mpc_signature, "message": challenge,
    })
    return verify_response.json().get("valid", False)

challenge = "random-challenge-12345"
is_valid = prove_mpc_wallet_control("did:msg:agent:alice", "mpc-wallet-uuid-abc123", challenge, "mpc-sig...")
print(f"Proof valid: {is_valid}")

6.4 智能合约 DID 解析与 MPC 验证

def resolve_and_verify_mpc(did: str, message: str, signature: str) -> dict:
    response = client.post("/agent/v1/query/contract", json={
        "address": DID_REGISTRY_ADDRESS,
        "query": {"resolve_and_verify": {"did": did, "message": message, "signature": signature, "algorithm": "Dilithium5"}},
    })
    response.raise_for_status()
    return response.json()

def query_mpc_wallet_verification(contract_address: str, wallet_id: str, did: str) -> dict:
    response = client.post("/agent/v1/query/contract", json={
        "address": contract_address,
        "query": {"get_wallet_verification": {"wallet_id": wallet_id, "did": did}},
    })
    response.raise_for_status()
    return response.json()

七、密钥分片备份与恢复

7.1 备份原则

7.2 分片备份格式

import json
import base64
import os
import hashlib
from datetime import datetime
from dataclasses import dataclass
from typing import Optional

try:
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    HAS_CRYPTO = True
except ImportError:
    HAS_CRYPTO = False
    print("Install cryptography: pip install cryptography")


@dataclass
class ShardBackup:
    backup_version: str = "1.0"
    wallet_id: str = ""
    wallet_address: str = ""
    signer_did: str = ""
    shard_index: int = 0
    total_shards: int = 0
    threshold: int = 0
    encrypted_shard: str = ""
    encryption_algorithm: str = "AES-256-GCM"
    salt: str = ""
    kdf: str = "PBKDF2-HMAC-SHA256"
    kdf_iterations: int = 600_000
    created_at: str = ""
    chain_id: str = "msg-chain-1"
    shard_hash: str = ""
    backup_hash: str = ""

    def to_json(self) -> str:
        data = {
            "backup_version": self.backup_version, "wallet_id": self.wallet_id,
            "wallet_address": self.wallet_address, "signer_did": self.signer_did,
            "shard_index": self.shard_index, "total_shards": self.total_shards,
            "threshold": self.threshold, "encrypted_shard": self.encrypted_shard,
            "encryption_algorithm": self.encryption_algorithm, "salt": self.salt,
            "kdf": self.kdf, "kdf_iterations": self.kdf_iterations,
            "created_at": self.created_at, "chain_id": self.chain_id,
            "shard_hash": self.shard_hash,
        }
        data["backup_hash"] = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
        return json.dumps(data, indent=2, ensure_ascii=False)

    @classmethod
    def from_json(cls, json_str: str) -> "ShardBackup":
        data = json.loads(json_str)
        stored_hash = data.pop("backup_hash", "")
        computed_hash = hashlib.sha256(json.dumps(data, sort_keys=True).encode()).hexdigest()
        if stored_hash and stored_hash != computed_hash:
            raise ValueError("Backup integrity check failed — tampered")
        return cls(**data)

    @classmethod
    def create(cls, wallet_id: str, wallet_address: str, signer_did: str,
               shard_index: int, total_shards: int, threshold: int,
               raw_shard: bytes, passphrase: str) -> "ShardBackup":
        salt = os.urandom(32)
        key = hashlib.pbkdf2_hmac("sha256", passphrase.encode("utf-8"), salt, 600_000, dklen=32)

        if HAS_CRYPTO:
            aesgcm = AESGCM(key)
            nonce = os.urandom(12)
            ciphertext = aesgcm.encrypt(nonce, raw_shard, None)
            encrypted_data = base64.b64encode(nonce + ciphertext).decode("ascii")
        else:
            from cryptography.hazmat.primitives.ciphers.aead import AESGCM
            aesgcm = AESGCM(key)
            nonce = os.urandom(12)
            ciphertext = aesgcm.encrypt(nonce, raw_shard, None)
            encrypted_data = base64.b64encode(nonce + ciphertext).decode("ascii")

        return cls(
            wallet_id=wallet_id, wallet_address=wallet_address, signer_did=signer_did,
            shard_index=shard_index, total_shards=total_shards, threshold=threshold,
            encrypted_shard=encrypted_data, salt=base64.b64encode(salt).decode("ascii"),
            created_at=datetime.utcnow().isoformat() + "Z",
            shard_hash=hashlib.sha256(raw_shard).hexdigest(),
        )


def backup_shard_to_file(shard_backup: ShardBackup, filepath: str):
    with open(filepath, "w") as f:
        f.write(shard_backup.to_json())
    print(f"Backup written to {filepath}")


def recover_shard_from_backup(backup_path: str, passphrase: str) -> bytes:
    with open(backup_path, "r") as f:
        backup = ShardBackup.from_json(f.read())
    salt = base64.b64decode(backup.salt)
    key = hashlib.pbkdf2_hmac("sha256", passphrase.encode("utf-8"), salt, backup.kdf_iterations, dklen=32)
    encrypted = base64.b64decode(backup.encrypted_shard)
    nonce = encrypted[:12]
    ciphertext = encrypted[12:]
    from cryptography.hazmat.primitives.ciphers.aead import AESGCM
    aesgcm = AESGCM(key)
    raw_shard = aesgcm.decrypt(nonce, ciphertext, None)
    recovered_hash = hashlib.sha256(raw_shard).hexdigest()
    if recovered_hash != backup.shard_hash:
        raise ValueError("Shard integrity check failed")
    print(f"Shard recovered: {backup.signer_did} index {backup.shard_index}")
    return raw_shard

7.3 签名者丢失处理

def handle_lost_signer(wallet_id: str, lost_signer_did: str, remaining_signers: list[dict], new_threshold=None) -> dict:
    print(f"Removing lost signer: {lost_signer_did}")
    remove_result = remove_mpc_signer(wallet_id, lost_signer_did)
    remaining_count = remove_result.get("total_signers", 0)
    current_threshold = remove_result.get("threshold", 1)
    if remaining_count < current_threshold:
        print(f"WARNING: {remaining_count} < {current_threshold} — wallet LOCKED")
    if remaining_count >= current_threshold:
        re_shard_result = trigger_re_shard(wallet_id=wallet_id, new_threshold=new_threshold or current_threshold)
        return re_shard_result
    return remove_result


def trigger_re_shard(wallet_id: str, new_threshold: int) -> dict:
    response = client.post(f"/agent/v1/mpc/wallet/{wallet_id}/reshard", json={"new_threshold": new_threshold})
    if response.headers.get("X-MSG-Stub") == "true":
        print("[Stub] Re-shard endpoint")
    response.raise_for_status()
    return response.json()


def dr_scenario_check(wallet_id: str) -> dict:
    wallet = query_mpc_wallet(wallet_id)
    total = wallet["total_signers"]
    threshold = wallet["threshold"]
    active = sum(1 for s in wallet["signers"] if s["status"] == "active")
    print(f"Active: {active}, Threshold: {threshold}, Can sign: {active >= threshold}")
    return {
        "wallet_id": wallet_id, "active": active, "threshold": threshold,
        "can_sign": active >= threshold, "max_loss_tolerated": active - threshold,
        "status": "OK" if active >= threshold else "LOCKED",
    }

7.4 灾难恢复剧本

场景 1：单个签名者丢失密钥分片

1. 移除丢失签名者 → 2. 检查 (n-1) >= t ? → 3. 可选添加新签名者 → 4. re-shard → 5. 测试签名 → 6. 更新备份

场景 2：多个签名者同时丢失（≥ n - t + 1）

1. 钱包锁定 → 2. 从备份恢复分片 → 3. 重建私钥 → 4. 创建新钱包转移资金

场景 3：备份存储泄露

1. 假设所有备份泄露 → 2. 冻结钱包 → 3. 用剩余签名者转移资金 → 4. 重建新钱包

八、安全最佳实践

8.1 阈值选择指南

8.2 签名者多样性

signer_diversity:
  - dimension: "地理"       ; requirement: "不同数据中心或云区域"
  - dimension: "网络"       ; requirement: "不同 ISP 和网络路径"
  - dimension: "基础设施"   ; requirement: "不同提供商或自托管"
  - dimension: "操作系统"   ; requirement: "Linux, macOS, BSD 混用"
  - dimension: "密钥存储"   ; requirement: "不同 HSM 或软件钱包"

8.3 会话超时调优

def calculate_session_timeout(num_signers: int, signing_method: str = "sequential", max_network_latency_ms: int = 500) -> int:
    base_time = 30
    per_signer_time = 60 if signing_method == "sequential" else 30
    network_buffer = (max_network_latency_ms / 1000) * num_signers * 2
    total = base_time + (per_signer_time * num_signers) + network_buffer
    return max(((int(total) + 29) // 30) * 30, 60)

print(f"2 sequential: {calculate_session_timeout(2, 'sequential')}s")
print(f"3 parallel: {calculate_session_timeout(3, 'parallel')}s")

8.4 防重放保护

def build_mpc_payload_with_replay_protection(tx_body: dict, chain_id: str, account_number: int, sequence: int, nonce=None) -> dict:
    import uuid
    return {"tx_body": tx_body, "chain_id": chain_id, "account_number": account_number,
            "sequence": sequence, "nonce": nonce or uuid.uuid4().hex, "created_at": int(time.time())}

def verify_replay_protection(signed_payload: dict, expected_chain_id: str, used_nonces: set) -> bool:
    if signed_payload.get("chain_id") != expected_chain_id:
        return False
    nonce = signed_payload.get("nonce", "")
    if nonce in used_nonces:
        return False
    used_nonces.add(nonce)
    if time.time() - signed_payload.get("created_at", 0) > 300:
        return False
    return True

8.5 审计日志要求

MPC_AUDIT_EVENTS = [
    "mpc_wallet_created", "mpc_wallet_queried", "mpc_signer_added",
    "mpc_signer_removed", "mpc_signer_lost_reported", "mpc_session_initiated",
    "mpc_partial_sig_submitted", "mpc_partial_sig_invalid", "mpc_session_completed",
    "mpc_session_failed", "mpc_session_timeout", "mpc_session_aborted",
    "mpc_shard_backup_created", "mpc_shard_recovered", "mpc_reshard_triggered",
    "mpc_tx_broadcast", "mpc_tx_confirmed",
]

def write_mpc_audit_log(event_type: str, wallet_id: str, status: str, **extra_fields):
    import uuid
    entry = {
        "event_id": uuid.uuid4().hex, "event_type": event_type,
        "timestamp": datetime.utcnow().isoformat() + "Z",
        "wallet_id": wallet_id, "status": status, **extra_fields,
    }
    print(f"[MPC AUDIT] {json.dumps(entry, ensure_ascii=False)}")
    return entry["event_id"]

8.6 本地沙箱测试

# Start local MSG Chain sandbox
./bin/quantum_node_linux start --sandbox

# Create test DIDs
curl -X POST http://localhost:8080/agent/v1/registry/register \
  -H "X-API-Key: test-key" \
  -d '{"agent_id": "test-agent-alice", ...}'

# Create MPC wallet
curl -X POST http://localhost:8080/agent/v1/mpc/wallet \
  -H "X-API-Key: test-key" -d '{...}'

# Initiate test signing
curl -X POST http://localhost:8080/agent/v1/mpc/wallet/{id}/sign \
  -H "X-API-Key: test-key" -d '{...}'

测试清单：

MPC_TEST_CHECKLIST = [
    "Create 2-of-3 MPC wallet",
    "Query wallet by ID",
    "Initiate signing session",
    "Submit partial signature as signer 1",
    "Submit partial signature as signer 2",
    "Session reaches COMPLETED",
    "Full signature is valid Dilithium-5 format",
    "Add a new signer",
    "Remove a signer",
    "Sign with new signer set",
    "Session correctly fails with insufficient signatures",
    "Session correctly times out",
    "Recover shard from backup",
    "Verify replay protection works",
]

九、完整示例项目

9.1 项目结构

mpc-agent-example/
├── config/
│   ├── signers.yaml
│   └── wallet.yaml
├── contracts/
│   └── mpc_accepting_contract/
│       ├── src/
│       │   ├── contract.rs
│       │   ├── msg.rs
│       │   └── state.rs
│       └── Cargo.toml
├── scripts/
│   ├── setup.py
│   ├── coordination.ts
│   └── verify.py
└── docker-compose.yml

9.2 CosmWasm 合约：接受 MPC 签名的消息

// contracts/mpc_accepting_contract/src/msg.rs

use cosmwasm_schema::{cw_serde, QueryResponsive};
use cosmwasm_std::Binary;

#[cw_serde]
pub struct InstantiateMsg {
    pub admin: String,
    pub allowed_wallet: Option<String>,
}

#[cw_serde]
pub enum ExecuteMsg {
    MpcSignedAction {
        action: String,
        params: String,
        signature: Binary,
        signer_did: String,
        nonce: String,
    },
    AdminAction {
        action: String,
        params: String,
    },
}

#[cw_serde]
#[derive(QueryResponsive)]
pub enum QueryMsg {
    GetState {},
    VerifyMpcSignature { message: String, signature: Binary, wallet_id: String },
    GetAllowedWallet {},
}

#[cw_serde]
pub struct StateResponse {
    pub action_count: u64,
    pub last_action: String,
    pub last_signer: String,
    pub admin: String,
}

#[cw_serde]
pub struct AllowedWalletResponse {
    pub wallet_id: Option<String>,
    pub wallet_address: Option<String>,
}

// contracts/mpc_accepting_contract/src/state.rs

use cosmwasm_schema::cw_serde;
use cw_storage_plus::{Item, Map};

#[cw_serde]
pub struct State {
    pub action_count: u64,
    pub admin: String,
}

#[cw_serde]
pub struct AllowedMpcWallet {
    pub wallet_id: String,
    pub wallet_address: String,
    pub aggregated_pubkey: String,
}

pub const STATE: Item<State> = Item::new("state");
pub const ALLOWED_WALLET: Item<AllowedMpcWallet> = Item::new("allowed_wallet");
pub const USED_NONCES: Map<String, u64> = Map::new("used_nonces");

// contracts/mpc_accepting_contract/src/contract.rs

use cosmwasm_std::{
    entry_point, to_binary, Binary, Deps, DepsMut, Env, MessageInfo,
    Response, StdError, StdResult,
};
use crate::msg::{
    AllowedWalletResponse, ExecuteMsg, InstantiateMsg, QueryMsg, StateResponse,
};
use crate::state::{AllowedMpcWallet, State, ALLOWED_WALLET, STATE, USED_NONCES};

#[entry_point]
pub fn instantiate(deps: DepsMut, _env: Env, info: MessageInfo, msg: InstantiateMsg) -> StdResult<Response> {
    let state = State { action_count: 0, admin: msg.admin.clone() };
    STATE.save(deps.storage, &state)?;
    if let Some(wallet_id) = msg.allowed_wallet {
        let wallet = AllowedMpcWallet {
            wallet_id,
            wallet_address: String::new(),
            aggregated_pubkey: String::new(),
        };
        ALLOWED_WALLET.save(deps.storage, &wallet)?;
    }
    Ok(Response::new().add_attribute("method", "instantiate").add_attribute("admin", msg.admin))
}

#[entry_point]
pub fn execute(deps: DepsMut, env: Env, info: MessageInfo, msg: ExecuteMsg) -> StdResult<Response> {
    match msg {
        ExecuteMsg::MpcSignedAction { action, params, signature, signer_did, nonce } => {
            execute_mpc_signed(deps, env, info, action, params, signature, signer_did, nonce)
        }
        ExecuteMsg::AdminAction { action, params } => execute_admin(deps, env, info, action, params),
    }
}

fn execute_mpc_signed(
    deps: DepsMut, _env: Env, _info: MessageInfo,
    action: String, _params: String, _signature: Binary, _signer_did: String, nonce: String,
) -> StdResult<Response> {
    if USED_NONCES.has(deps.storage, nonce.clone()) {
        return Err(StdError::generic_err("Nonce already used"));
    }
    let _wallet = ALLOWED_WALLET.load(deps.storage)?;
    // In production: verify Dilithium-5 signature against wallet.aggregated_pubkey
    USED_NONCES.save(deps.storage, nonce.clone(), &_env.block.height)?;

    let mut state = STATE.load(deps.storage)?;
    state.action_count += 1;
    state.last_action = action.clone();
    STATE.save(deps.storage, &state)?;

    Ok(Response::new()
        .add_attribute("method", "mpc_signed_action")
        .add_attribute("action", action)
        .add_attribute("signer", _signer_did)
        .add_attribute("count", state.action_count.to_string()))
}

fn execute_admin(deps: DepsMut, _env: Env, info: MessageInfo, action: String, _params: String) -> StdResult<Response> {
    let state = STATE.load(deps.storage)?;
    if info.sender != state.admin {
        return Err(StdError::generic_err("Unauthorized: admin only"));
    }
    Ok(Response::new().add_attribute("method", "admin_action").add_attribute("action", action))
}

#[entry_point]
pub fn query(deps: Deps, _env: Env, msg: QueryMsg) -> StdResult<Binary> {
    match msg {
        QueryMsg::GetState {} => {
            let state = STATE.load(deps.storage)?;
            to_binary(&StateResponse {
                action_count: state.action_count,
                last_action: String::new(),
                last_signer: String::new(),
                admin: state.admin,
            })
        }
        QueryMsg::VerifyMpcSignature { message: _, signature: _, wallet_id: _ } => {
            to_binary(&AllowedWalletResponse { wallet_id: None, wallet_address: None })
        }
        QueryMsg::GetAllowedWallet {} => {
            let wallet = ALLOWED_WALLET.load(deps.storage)?;
            to_binary(&AllowedWalletResponse {
                wallet_id: Some(wallet.wallet_id),
                wallet_address: Some(wallet.wallet_address),
            })
        }
    }
}

9.3 TypeScript 代理协调脚本

import { SigningCosmWasmClient, CosmWasmClient } from "@cosmjs/cosmwasm-stargate";
import { DirectSecp256k1HdWallet } from "@cosmjs/proto-signing";
import { GasPrice } from "@cosmjs/stargate";

const CONFIG = {
  apiBase: "http://localhost:8080",
  rpcEndpoint: "http://localhost:26657",
  chainId: "msg-chain-1",
  contractAddress: "msg1...mpcAcceptingContract",
};

const SIGNERS = [
  { did: "did:msg:agent:alice", apiKey: "alice-api-key", keyShare: new Uint8Array(32).fill(1) },
  { did: "did:msg:agent:bob", apiKey: "bob-api-key", keyShare: new Uint8Array(32).fill(2) },
  { did: "did:msg:agent:carol", apiKey: "carol-api-key", keyShare: new Uint8Array(32).fill(3) },
];

async function setupEnvironment() {
  console.log("Creating MPC wallet (2-of-3)...");
  const walletResponse = await fetch(`${CONFIG.apiBase}/agent/v1/mpc/wallet`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "X-API-Key": SIGNERS[0].apiKey },
    body: JSON.stringify({
      name: "agent-coordination-wallet",
      threshold: 2,
      total_signers: 3,
      signers: SIGNERS.map(s => ({ did: s.did, pubkey: `dilithium5:${Buffer.from(s.keyShare).toString("hex")}` })),
      chain_id: CONFIG.chainId,
    }),
  });
  const wallet = await walletResponse.json();
  console.log(`Wallet ID: ${wallet.wallet_id}, Address: ${wallet.address}`);
  return wallet;
}

async function initAndSign(walletId: string, message: string) {
  console.log("Initiating signing session...");
  const sessionRes = await fetch(`${CONFIG.apiBase}/agent/v1/mpc/wallet/${walletId}/sign`, {
    method: "POST",
    headers: { "Content-Type": "application/json", "X-API-Key": SIGNERS[0].apiKey },
    body: JSON.stringify({
      payload: { tx_type: "custom", data: message },
      session_timeout: 300,
      signer_order: [SIGNERS[0].did, SIGNERS[1].did],
      chain_id: CONFIG.chainId,
    }),
  });
  const session = await sessionRes.json();
  console.log(`Session: ${session.session_id} (${session.status})`);

  for (const signer of [SIGNERS[0], SIGNERS[1]]) {
    const submitRes = await fetch(
      `${CONFIG.apiBase}/agent/v1/mpc/wallet/${walletId}/sign/${session.session_id}/submit`,
      {
        method: "POST",
        headers: { "Content-Type": "application/json", "X-API-Key": signer.apiKey },
        body: JSON.stringify({
          signer_did: signer.did,
          partial_signature: Buffer.from(signer.keyShare).toString("base64"),
          nonce: crypto.randomUUID(),
        }),
      }
    );
    console.log(`${signer.did}: ${(await submitRes.json()).status}`);
  }
  return session.session_id;
}

async function main() {
  const wallet = await setupEnvironment();
  const sessionId = await initAndSign(wallet.wallet_id, "MPC-signed action from AI Agent");
  console.log(`Complete! Session: ${sessionId}`);
}

main().catch(console.error);

9.4 Docker Compose 本地沙箱

version: "3.8"
services:
  msg-chain-node:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "26657:26657"
      - "1317:1317"
      - "8080:8080"
      - "9090:9090"
    volumes:
      - msg-chain-data:/root/.msg-chain
    command: ["./bin/quantum_node_linux", "start", "--sandbox"]
    environment:
      - MSG_CHAIN_ID=msg-chain-1
      - MSG_PREFIX=msg

volumes:
  msg-chain-data:

十、边界与限制

10.1 Stub 端点

Stub 端点检测：

def check_stub(response: httpx.Response) -> bool:
    return response.headers.get("X-MSG-Stub") == "true"

10.2 agent_mpc_v1 合约状态

• 实现程度： 部分实现
• 缺失功能： 实际阈值密码学计算、链上 MPC 签名验证、分片管理
• 当前能力： 查询接口可能工作，写操作返回 stub 响应
• 依赖： 底层的 Dilithium-5 签名已完全实现，但阈值扩展尚未完成

10.3 生产环境限制

1. 无公开主网： MSG Chain 目前没有公开主网。所有测试在本地沙箱进行。
2. 写操作不稳定： 所有写路径都标记为 stub，可能不持久化状态。
3. 阈值密码学未部署： 实际的 MPC 计算（部分签名生成、聚合）尚未在生产环境中实施。
4. 合约未实例化： agent_mpc_v1 可能在链上未部署或处于禁用状态。
5. 限流： 公开 GET 路径限流 100 req/s（突发 200），受保护 POST 路径限流 20 req/s（突发 40）。

10.4 限流处理

import time

def rate_limited_request(client: httpx.Client, method: str, path: str, **kwargs) -> httpx.Response:
    for attempt in range(3):
        response = client.request(method, path, **kwargs)
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", str(2 ** attempt)))
            print(f"Rate limited. Retrying in {retry_after}s...")
            time.sleep(retry_after)
            continue
        return response
    raise TimeoutError("Rate limit retries exhausted")

10.5 过渡计划

当 MSG Chain 完成 MPC 实现后，迁移路径：

当前（部分实现 / Stub）       未来（完全实现）
────────────────────────      ────────────────
POST .../mpc/wallet (Stub) →  POST .../mpc/wallet (真实)
agent_mpc_v1 (部分实现)    →  agent_mpc_v1 (完全实现)
模拟部分签名                 →  真实阈值 Dilithium-5 签名
手动分片备份                 →  自动分片管理与恢复
无生产环境                   →  公开主网

附录一：API 参考

MPC 端点汇总

链参数

端点地址

附录二：错误码表

MPC 相关错误码

错误响应格式

{
  "error": {
    "code": 3,
    "name": "UNAUTHORIZED",
    "message": "未授权：缺少 X-API-Key 请求头",
    "details": "写操作需要提供有效的 API Key。请在 HTTP 头中添加 X-API-Key。"
  },
  "request_id": "req-uuid-1234"
}

Stub 识别

async def is_stub(response: httpx.Response) -> bool:
    return response.headers.get("X-MSG-Stub") == "true"

附录三：日志格式

结构化日志 Schema

{
  "event_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "event_type": "mpc_session_completed",
  "timestamp": "2026-07-06T12:02:30Z",
  "wallet_id": "mpc-wallet-uuid-abc123",
  "session_id": "mpc-sess-uuid-xyz789",
  "signer_did": "did:msg:agent:alice",
  "client_ip": "192.168.1.100",
  "request_id": "req-abc-123",
  "status": "success",
  "error_code": 0,
  "error_message": null,
  "duration_ms": 150234,
  "threshold": 2,
  "signers_used": ["did:msg:agent:alice", "did:msg:agent:bob"],
  "signature_size_bytes": 4595,
  "payload_hash": "sha256:abc123def456..."
}

日志级别

Python 日志工厂

import logging
import json

class MPCLogger:
    def __init__(self, wallet_id: str):
        self.wallet_id = wallet_id
        self.logger = logging.getLogger(f"mpc.{wallet_id}")

    def log(self, event_type: str, status: str, **extra):
        import uuid
        entry = {
            "event_id": uuid.uuid4().hex,
            "event_type": event_type,
            "timestamp": datetime.utcnow().isoformat() + "Z",
            "wallet_id": self.wallet_id,
            "status": status,
            **extra,
        }
        self.logger.info(json.dumps(entry, ensure_ascii=False))

    def audit(self, event_type: str, **extra):
        self.log(event_type, "audit", severity="AUDIT", **extra)

文档版本： v1.0.0
相关文档： MSG 链 AI Agent 开发框架 | Dilithium-5 后量子签名开发者指南 | API 接口大全