MSG Chain AI Agent OpenAPI 能力表面与端点指南
链 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 参数，永不自主铸造/销毁代币
核心原则: 接口表面真实存在，但能力成熟度分层明显。
本文档忠实反映partial(部分实现) 状态，不宣称生产就绪。
目录
1. 概述
1.1 什么是 Agent API Surface
MSG Chain 的 Agent API 能力表面 (Agent API Surface) 是白皮书系统中
将 Agent 的可调用能力从底层控制面中单独拆出来的一层接口抽象。
它在白皮书知识网络中的定位:
- 分组: 开发者与接口 (
developer) - 状态: 部分实现 (
partial) - 出链:
ai_agent.html,ai_control_plane.html,ai_policy_capability.html,
rpc.html,sdk_dev_surface.html,trust_verdict.html - 回链:
ai_task_l2.html,l1_atomic_modular.html,sdk_dev_surface.html,
overview.html
1.2 真实位置
OpenAPI 规范文件: docs/api/agent_openapi.yaml
OpenAPI 公共导出: whitepaper/api_specs/openapi/agent_surface.yaml
模块 HTML: whitepaper/modules/agent_api_surface.html
机器导出 JSON: whitepaper/module_exports/agent_api_surface.json
1.3 能力成熟度矩阵总览
白皮书系统通过 developer_capability_matrix.json 定义了 16 个能力表面，
每个表面有 6 个核心字段:
# 核心字段说明
fields = {
'surface_id': '表面的唯一标识符',
'machine_readiness': '机器就绪度评级',
'machine_safe_for_codegen': 'AI 是否可以直接据此生成代码',
'write_path_ready': '写路径是否可用',
'production_supported': '是否支持生产环境使用',
}
# 机器就绪度评级体系（从高到低）
READINESS_LEVELS = [
'production_reference', # 开发参考级参考 — 可安全用于生产寻址
'assisted_codegen', # 辅助代码生成 — 可安全生成代码，有写路径
'starter_ready', # 启动器可用 — 可安全用于快速启动
'source_backed_reference', # 源码参考 — 可读但写路径不可用
'guarded_integration', # 受保护集成 — 写路径可用但非生产
'guarded_write', # 受保护写 — 不可自动执行，需门禁
'guarded_release', # 受保护发布 — 不可自动执行
'guarded_execution', # 受保护执行 — 可理解但不可自动执行
'read_only_assist', # 只读辅助 — 仅可读
'local_candidate', # 本地候选 — alpha 级，不可用于生产
'fail_closed_reference', # 关闭态参考 — 明确不可用
]
2. OpenAPI 端点分组
2.1 分组全景
当前 docs/api/agent_openapi.yaml 中可见的端点分组:
from enum import Enum
from typing import Dict, List, Optional
class EndpointStatus(str, Enum):
IMPLEMENTED = 'implemented' # 已实现，可直接使用
PARTIAL = 'partial' # 部分实现，功能有限
STUB = 'stub' # 桩端点，返回 X-MSG-Stub: true
class AuthRequirement(str, Enum):
PUBLIC = 'public'
API_KEY = 'API Key required'
GUARDED = 'guarded: human_approval_gates'
STUB_NA = 'N/A (stub)'
class APIGroup:
def __init__(
self,
name: str,
status: EndpointStatus,
endpoints: List[str],
description: str,
auth: AuthRequirement,
):
self.name = name
self.status = status
self.endpoints = endpoints
self.description = description
self.auth = auth
def is_stub(self) -> bool:
return self.status == EndpointStatus.STUB
def needs_api_key(self) -> bool:
return self.auth in (AuthRequirement.API_KEY, AuthRequirement.GUARDED)
def summary(self) -> str:
return (
f'[{self.status.value}] {self.name}: '
f'{", ".join(self.endpoints) if self.endpoints else "—"} '
f'| Auth: {self.auth.value}'
)
API_GROUPS: Dict[str, APIGroup] = {
'Query': APIGroup(
name='Query',
status=EndpointStatus.PARTIAL,
endpoints=['account', 'balance', 'tx', 'block', 'batch'],
description='查询类端点，已具真实读取面，适合 AI preflight 检查',
auth=AuthRequirement.PUBLIC,
),
'Wallet': APIGroup(
name='Wallet',
status=EndpointStatus.PARTIAL,
endpoints=['transfer', 'sign', 'broadcast'],
description='写操作端点，需 API Key 认证',
auth=AuthRequirement.API_KEY,
),
'NFT': APIGroup(
name='NFT',
status=EndpointStatus.PARTIAL,
endpoints=['query', 'transfer', 'metadata'],
description='NFT 查询与操作 — 部分实现',
auth=AuthRequirement.GUARDED,
),
'Events': APIGroup(
name='Events',
status=EndpointStatus.PARTIAL,
endpoints=['history', 'subscribe'],
description='事件历史与订阅 — history 公开，subscribe 需 API Key',
auth=AuthRequirement.PUBLIC,
# 注: history 公开; subscribe 需 API Key
),
'Registry': APIGroup(
name='Registry',
status=EndpointStatus.PARTIAL,
endpoints=['register', 'discover', 'resolve'],
description='Agent 注册与发现表面 — 写操作需 API Key',
auth=AuthRequirement.API_KEY,
),
'MPC': APIGroup(
name='MPC',
status=EndpointStatus.PARTIAL,
endpoints=['sign', 'keygen', 'aggregate'],
description='多签与签名编排 — 需 API Key',
auth=AuthRequirement.API_KEY,
),
'DeFi / Oracle / Payment / Bridge': APIGroup(
name='DeFi / Oracle / Payment / Bridge',
status=EndpointStatus.STUB,
endpoints=[],
description='返回 X-MSG-Stub: true — 未实现',
auth=AuthRequirement.STUB_NA,
),
}
def list_all_groups() -> None:
"""打印所有 API 分组的状态摘要。"""
for key, group in API_GROUPS.items():
print(group.summary())
def get_active_endpoints() -> Dict[str, List[str]]:
"""获取非 Stub 分组中的端点列表。"""
return {
g.name: g.endpoints
for g in API_GROUPS.values()
if not g.is_stub()
}
def get_stub_groups() -> List[str]:
"""获取所有 Stub 分组名称。"""
return [g.name for g in API_GROUPS.values() if g.is_stub()]
2.2 Query 组 — 真实可用
Query 组是当前最成熟的分组。所有端点公开可访问，无需 API Key。
import httpx
from typing import Optional, Dict, Any
MSG_AGENT_BASE = 'https://msgchain.org/agent/v1'
class QueryClient:
"""Query 组客户端 — 无需 API Key。"""
def __init__(self, base_url: str = MSG_AGENT_BASE):
self.base_url = base_url
self._client = httpx.AsyncClient(timeout=30.0)
async def account(self, address: str) -> Dict[str, Any]:
"""查询账户摘要。"""
resp = await self._client.get(
f'{self.base_url}/query/account',
params={'address': address},
)
resp.raise_for_status()
return resp.json()
async def balance(
self, account: str, asset: str
) -> Dict[str, Any]:
"""查询单资产余额。
返回示例 (来自 agent_surface.yaml):
{
"account": "msg1developerexample...",
"asset": "umsg",
"amount": "125000000",
"height": 128
}
"""
resp = await self._client.get(
f'{self.base_url}/query/balance',
params={'account': account, 'asset': asset},
)
resp.raise_for_status()
return resp.json()
async def tx(self, tx_hash: str) -> Dict[str, Any]:
"""按哈希查询交易状态。"""
resp = await self._client.get(
f'{self.base_url}/query/tx',
params={'hash': tx_hash},
)
resp.raise_for_status()
return resp.json()
async def block(self, height: int) -> Dict[str, Any]:
"""查询区块详情。"""
resp = await self._client.get(
f'{self.base_url}/query/block',
params={'height': height},
)
resp.raise_for_status()
return resp.json()
async def batch(
self, queries: List[Dict[str, Any]]
) -> List[Dict[str, Any]]:
"""批量查询。"""
resp = await self._client.post(
f'{self.base_url}/query/batch',
json={'queries': queries},
)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self._client.aclose()
使用示例:
async def preflight_check(address: str) -> None:
"""AI preflight 检查 — 构建交易前的余额确认。"""
client = QueryClient()
try:
acct = await client.account(address)
bal = await client.balance(address, 'umsg')
print(f'账户: {acct}')
print(f'余额: {bal["amount"]} {bal["asset"]} @ height {bal["height"]}')
finally:
await client.close()
2.3 Wallet 组 — 需 API Key
Wallet 组执行写操作，必须提供 API Key。
class WalletClient:
"""Wallet 组客户端 — 需 API Key。"""
def __init__(
self,
api_key: str,
base_url: str = MSG_AGENT_BASE,
):
self.base_url = base_url
self._headers = {'X-API-Key': api_key}
self._client = httpx.AsyncClient(
headers=self._headers, timeout=30.0
)
async def transfer(
self,
from_addr: str,
to_addr: str,
amount: str,
denom: str = 'umsg',
memo: str = '',
) -> Dict[str, Any]:
"""发起转账。"""
resp = await self._client.post(
f'{self.base_url}/wallet/transfer',
json={
'from': from_addr,
'to': to_addr,
'amount': amount,
'denom': denom,
'memo': memo,
},
)
resp.raise_for_status()
return resp.json()
async def sign(
self, account: str, unsigned_payload: Dict[str, Any]
) -> Dict[str, Any]:
"""签名交易。"""
resp = await self._client.post(
f'{self.base_url}/wallet/sign',
json={
'account': account,
'unsigned_payload': unsigned_payload,
},
)
resp.raise_for_status()
return resp.json()
async def broadcast(self, signed_tx_hex: str) -> Dict[str, Any]:
"""广播已签名交易。"""
resp = await self._client.post(
f'{self.base_url}/wallet/broadcast',
json={'signed_tx_hex': signed_tx_hex},
)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self._client.aclose()
2.4 Events 组 — 混合认证
class EventsClient:
"""Events 组客户端。"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = MSG_AGENT_BASE,
):
self.base_url = base_url
headers = {}
if api_key:
headers['X-API-Key'] = api_key
self._client = httpx.AsyncClient(headers=headers, timeout=30.0)
self._api_key = api_key
async def history(
self,
from_height: Optional[int] = None,
limit: int = 100,
) -> Dict[str, Any]:
"""查询历史事件 — 公开。"""
params = {'limit': limit}
if from_height is not None:
params['from_height'] = from_height
resp = await self._client.get(
f'{self.base_url}/events/history',
params=params,
)
resp.raise_for_status()
return resp.json()
async def subscribe(
self, topics: List[str], callback_url: str
) -> Dict[str, Any]:
"""订阅事件 — 需 API Key。"""
if not self._api_key:
raise PermissionError(
'subscribe 端点需要 API Key，请提供 api_key'
)
resp = await self._client.post(
f'{self.base_url}/events/subscribe',
json={
'topics': topics,
'callback_url': callback_url,
},
)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self._client.aclose()
2.5 Registry 组 — Agent 注册与发现
class RegistryClient:
"""Registry 组客户端。"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = MSG_AGENT_BASE,
):
self.base_url = base_url
headers = {}
if api_key:
headers['X-API-Key'] = api_key
self._client = httpx.AsyncClient(headers=headers, timeout=30.0)
self._api_key = api_key
async def resolve(self, agent_id: str) -> Dict[str, Any]:
"""解析 Agent 地址 — 公开。"""
resp = await self._client.get(
f'{self.base_url}/registry/resolve',
params={'agent_id': agent_id},
)
resp.raise_for_status()
return resp.json()
async def discover(
self, tags: Optional[List[str]] = None
) -> Dict[str, Any]:
"""发现 Agent — 公开。"""
params = {}
if tags:
params['tags'] = ','.join(tags)
resp = await self._client.get(
f'{self.base_url}/registry/discover',
params=params,
)
resp.raise_for_status()
return resp.json()
async def register(
self,
agent_id: str,
endpoint: str,
metadata: Dict[str, Any],
) -> Dict[str, Any]:
"""注册 Agent — 需 API Key。"""
if not self._api_key:
raise PermissionError(
'register 端点需要 API Key'
)
resp = await self._client.post(
f'{self.base_url}/registry/register',
json={
'agent_id': agent_id,
'endpoint': endpoint,
'metadata': metadata,
},
)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self._client.aclose()
2.6 MPC 组 — 多签与签名编排
class MPCClient:
"""MPC 组客户端 — 需 API Key。"""
def __init__(self, api_key: str, base_url: str = MSG_AGENT_BASE):
self.base_url = base_url
self._client = httpx.AsyncClient(
headers={'X-API-Key': api_key}, timeout=30.0
)
async def keygen(
self, party_count: int, threshold: int
) -> Dict[str, Any]:
"""生成 MPC 密钥。"""
resp = await self._client.post(
f'{self.base_url}/mpc/keygen',
json={
'party_count': party_count,
'threshold': threshold,
},
)
resp.raise_for_status()
return resp.json()
async def sign(
self,
session_id: str,
payload: Dict[str, Any],
) -> Dict[str, Any]:
"""MPC 签名。"""
resp = await self._client.post(
f'{self.base_url}/mpc/sign',
json={
'session_id': session_id,
'payload': payload,
},
)
resp.raise_for_status()
return resp.json()
async def aggregate(
self, signatures: List[Dict[str, Any]]
) -> Dict[str, Any]:
"""聚合签名。"""
resp = await self._client.post(
f'{self.base_url}/mpc/aggregate',
json={'signatures': signatures},
)
resp.raise_for_status()
return resp.json()
async def close(self) -> None:
await self._client.aclose()
2.7 端点认证矩阵速查
ENDPOINT_AUTH_MATRIX = {
# Query 组 — 全部公开
'GET /agent/v1/query/account': 'public',
'GET /agent/v1/query/balance': 'public',
'GET /agent/v1/query/tx': 'public',
'GET /agent/v1/query/block': 'public',
'POST /agent/v1/query/batch': 'public',
# Wallet 组 — 全部需 API Key
'POST /agent/v1/wallet/transfer': 'API Key required',
'POST /agent/v1/wallet/sign': 'API Key required',
'POST /agent/v1/wallet/broadcast':'API Key required',
# Events 组 — 混合
'GET /agent/v1/events/history': 'public',
'POST /agent/v1/events/subscribe':'API Key required',
# Registry 组 — 混合
'GET /agent/v1/registry/resolve': 'public',
'GET /agent/v1/registry/discover':'public',
'POST /agent/v1/registry/register':'API Key required',
# MPC 组 — 全部需 API Key
'POST /agent/v1/mpc/keygen': 'API Key required',
'POST /agent/v1/mpc/sign': 'API Key required',
'POST /agent/v1/mpc/aggregate': 'API Key required',
# Stub 组 — 无需认证（返回 stub）
'POST /agent/v1/defi/*': 'N/A (stub)',
'POST /agent/v1/oracle/*': 'N/A (stub)',
'POST /agent/v1/payment/*': 'N/A (stub)',
'POST /agent/v1/bridge/*': 'N/A (stub)',
}
def needs_api_key(endpoint_path: str) -> bool:
"""判断端点是否需要 API Key。"""
for path, auth in ENDPOINT_AUTH_MATRIX.items():
if path.endswith(endpoint_path) or endpoint_path in path:
return auth == 'API Key required'
return False
3. 能力成熟度矩阵
3.1 完整 16 表面矩阵
以下数据直接来自 whitepaper/developer_capability_matrix.json。
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class CapabilitySurface:
surface_id: str
title: str
machine_readiness: str
machine_safe_for_codegen: bool
write_path_ready: bool
schema_available: bool
example_available: bool
production_supported: bool
best_for: List[str] = field(default_factory=list)
blocking_gaps: List[str] = field(default_factory=list)
boundaries: List[str] = field(default_factory=list)
example_entry_points: List[str] = field(default_factory=list)
@property
def can_autogen(self) -> bool:
"""AI 是否可以自动生成代码。"""
return self.machine_safe_for_codegen
@property
def can_write(self) -> bool:
"""写路径是否可用。"""
return self.write_path_ready
@property
def is_production(self) -> bool:
"""是否生产就绪。"""
return self.production_supported
def readiness_emoji(self) -> str:
if self.is_production:
return '🟢'
elif self.can_write:
return '🟡'
elif self.can_autogen:
return '🟠'
else:
return '🔴'
def summary_line(self) -> str:
return (
f'{self.readiness_emoji()} {self.surface_id}: '
f'readiness={self.machine_readiness}, '
f'codegen={"✓" if self.can_autogen else "✗"}, '
f'write={"✓" if self.can_write else "✗"}, '
f'prod={"✓" if self.is_production else "✗"}'
)
CAPABILITY_MATRIX: List[CapabilitySurface] = [
CapabilitySurface(
surface_id='contract_runtime',
title='CosmWasm 合约运行时与生命周期',
machine_readiness='assisted_codegen',
machine_safe_for_codegen=True,
write_path_ready=True,
schema_available=True,
example_available=True,
production_supported=True,
best_for=[
'合约消息设计',
'状态流转设计',
'部署调用路径生成',
'回执验证规划',
],
blocking_gaps=[
'当前模板是 starter pack，不等于官方业务合约全集',
'缺少从链实现自动导出的正式 schema 契约',
'生产部署仍需真实权限与验收',
],
boundaries=[
'运行时主线已实现，且已补 starter template，但不等于所有业务模板都已完善。',
'AI 可辅助生成合约代码，但仍需结合业务规则、人类审核与真实部署账户完成落地。',
],
example_entry_points=[
'recipes/contract_minimal.json',
'contract_templates/index.json',
],
),
CapabilitySurface(
surface_id='core_contract_reference_pack',
title='核心合约消费索引与 interface/source pack',
machine_readiness='source_backed_reference',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'识别核心合约 canonical key 与 schema',
'生成 typed query/execute payload',
'定位 msg.rs / schema / Cargo 元数据',
],
blocking_gaps=[
'当前公开的是 interface/source 消费索引，不是全部核心合约逻辑公开包',
'signed schema release 与 canonical mapping cross-check 仍未关闭',
],
boundaries=[
'当前适合外部 AI 读取核心合约接口、schema 与 message source，'
'但不能误当成核心合约全量公开与 not independently verified for production ABI。',
'store/instantiate/migrate 等高风险动作仍必须服从治理、'
'审批与真实 receipt 证据。',
],
example_entry_points=[
'contract_reference/core_contracts.json',
'api_specs/formal_contracts.json',
],
),
CapabilitySurface(
surface_id='registry_resolution',
title='Registry Canonical Key 与地址解析',
machine_readiness='production_reference',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=False,
example_available=True,
production_supported=True,
best_for=[
'合约地址寻址',
'canonical key 解析',
'部署后地址发现',
],
blocking_gaps=[
'缺少独立 machine schema 定义 registry query 返回结构',
],
boundaries=[
'可作为机器寻址入口，但仍应以真实 query 返回为准。',
],
example_entry_points=[
'modules/registry.html',
'module_exports/registry.json',
],
),
CapabilitySurface(
surface_id='rpc_gateway',
title='RPC / REST / gRPC 查询与广播网关',
machine_readiness='assisted_codegen',
machine_safe_for_codegen=True,
write_path_ready=True,
schema_available=True,
example_available=True,
production_supported=True,
best_for=[
'dApp 查询层',
'交易广播',
'基础客户端封装',
],
blocking_gaps=[
'当前 OpenAPI 仍是文档流水线维护的协议摘要',
'live/public 端点仍需以真实可用性与证据为准',
],
boundaries=[
'当前已补机器可消费 OpenAPI 摘要，但仍不是从生产服务自动导出的最终 API 契约。',
],
example_entry_points=[
'api_specs/rpc_methods.json',
'api_specs/openapi/public_query.yaml',
],
),
CapabilitySurface(
surface_id='formal_api_schema_pack',
title='正式 API / Schema 契约索引',
machine_readiness='source_backed_reference',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'读取 repo-backed OpenAPI/Swagger/agent manifest',
'生成 request/response parser',
'统一 receipt/event/error code 消费入口',
],
blocking_gaps=[
'当前仍缺 signed schema release manifest',
'public route inventory、receipt consistency 与 negative suite '
'生产证据仍未关闭',
],
boundaries=[
'这层比白皮书摘要更接近真实源码与 release manifest，'
'但仍不能越级宣称 public production API 已完成。',
],
example_entry_points=[
'api_specs/formal_contracts.json',
'api_specs/openapi/agent_surface.yaml',
],
),
CapabilitySurface(
surface_id='wallet_frontend',
title='Keplr / CosmJS / 钱包前端接入',
machine_readiness='guarded_integration',
machine_safe_for_codegen=True,
write_path_ready=True,
schema_available=False,
example_available=True,
production_supported=False,
best_for=[
'dApp 钱包接入',
'chain config 注入',
'签名与查询前端路径',
],
blocking_gaps=[
'缺少对外稳定前端样例项目',
'兼容目标不等于生态全量上线',
],
boundaries=[
'当前是兼容路径与配置底座，不应表述成官方钱包生态已全量上线。',
],
example_entry_points=[
'recipes/dapp_minimal.json',
'chain_config/index.json',
],
),
CapabilitySurface(
surface_id='explorer_receipts',
title='Explorer 查询、receipt 与日志检索面',
machine_readiness='read_only_assist',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'receipt 回放',
'事件追踪',
'contract source 元数据定位',
],
blocking_gaps=[
'live/public Explorer 仍未全量关闭',
'索引与分页的生产 SLO/保留策略仍需独立交付',
],
boundaries=[
'可辅助开发调试与回证，不应直接当成完全生产化 Explorer 平台。',
],
example_entry_points=[
'modules/explorer.html',
'modules/indexer_data_plane.html',
],
),
CapabilitySurface(
surface_id='agent_query_and_guarded_write',
title='Agent API 查询面与受保护写路径',
machine_readiness='guarded_write',
machine_safe_for_codegen=False,
write_path_ready=False,
schema_available=False,
example_available=False,
production_supported=False,
best_for=[
'AI coding 辅助编排',
'监控读取',
'高风险动作门禁理解',
],
blocking_gaps=[
'部分写路径仍为 stub',
'缺少稳定 OpenAPI',
'自动执行仍需审批、风控、密钥与权限系统',
],
boundaries=[
'不能把 Agent 写路径当成已完成的全自动生产执行面。',
'所有高风险写动作仍需区分真实写路径、受保护路径与 stub 路径。',
],
example_entry_points=[
'api_specs/openapi/agent_surface.yaml',
'integration_examples/external_ai_agent_bootstrap_prompt.json',
],
),
CapabilitySurface(
surface_id='sdk_surface',
title='MSG SDK 开发表面',
machine_readiness='local_candidate',
machine_safe_for_codegen=False,
write_path_ready=False,
schema_available=False,
example_available=False,
production_supported=False,
best_for=[
'识别 alpha SDK 分层',
'为后续签名发布留出能力映射',
],
blocking_gaps=[
'尚未形成 signed/public SDK release',
'白皮书机器层仍不直接承载 SDK 安装证明',
],
boundaries=[
'SDK 已进入 alpha/local candidate，不能被 AI 误当成稳定公共客户端。',
],
example_entry_points=[
'modules/sdk_dev_surface.html',
'module_exports/sdk_dev_surface.json',
],
),
CapabilitySurface(
surface_id='chain_config_pack',
title='链配置与钱包注入包',
machine_readiness='starter_ready',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=True,
best_for=[
'前端链配置生成',
'Keplr suggestChain',
'CosmJS/钱包接入初始化',
],
blocking_gaps=[
'缺少从 live/public 节点自动探测并签名确认的网络元数据机制',
],
boundaries=[
'当前链配置来自仓库配置基线，生产接入前仍应复核 endpoint 可用性。',
],
example_entry_points=[
'chain_config/network_presets.json',
'chain_config/index.json',
],
),
CapabilitySurface(
surface_id='public_sandbox_strategy',
title='Public Devnet / Testnet / Faucet 策略',
machine_readiness='fail_closed_reference',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'判断 public sandbox 是否可用',
'决定走 public testnet 还是 local-only starter',
'避免把 admin/local faucet 误读成 public developer faucet',
],
blocking_gaps=[
'当前选择的是 Option B fail-closed 策略',
'signed disabled manifest、public faucet absence proof '
'与 final signatures 仍未关闭',
],
boundaries=[
'当前默认口径不是 public sandbox ready，而是 disabled/fail-closed + '
'明确 local development path。',
],
example_entry_points=[
'chain_config/developer_sandbox_strategy.json',
'quickstart/contract_and_dapp_minimal.json',
],
),
CapabilitySurface(
surface_id='contract_template_pack',
title='合约模板与 schema starter pack',
machine_readiness='starter_ready',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'快速生成最小可运行合约',
'补充消息 schema',
'生成 cargo test 骨架',
],
blocking_gaps=[
'模板仍是 starter，不覆盖所有业务模式',
'没有替代业务审计与测试',
],
boundaries=[
'模板包用于加速 AI coding，不等于官方系统合约或经过审计的业务模板。',
],
example_entry_points=[
'contract_templates/index.json',
'recipes/contract_minimal.json',
],
),
CapabilitySurface(
surface_id='dapp_starter_pack',
title='dApp Starter 与前端适配器样板',
machine_readiness='starter_ready',
machine_safe_for_codegen=True,
write_path_ready=True,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'快速搭建最小 dApp 前端',
'钱包连接与链配置样板',
'交易构造与广播 demo',
],
blocking_gaps=[
'样板不等于开发参考级 dApp',
'没有覆盖多语言/多钱包方案',
],
boundaries=[
'样板用于加速 dApp 开发原型，不应直接视作开发参考级前端。',
],
example_entry_points=[
'examples/index.json',
'recipes/dapp_minimal.json',
],
),
CapabilitySurface(
surface_id='developer_quickstart_pack',
title='开发者快速启动包',
machine_readiness='starter_ready',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'快速了解 MSG 开发流程',
'按推荐顺序消费机器入口',
],
blocking_gaps=[
'快速启动不等于完整上线路径',
],
boundaries=[
'用于引导开发流程，不能代替业务规则与人类审批。',
],
example_entry_points=[
'quickstart/contract_and_dapp_minimal.json',
'developer_entry.json',
],
),
CapabilitySurface(
surface_id='release_pack',
title='发布包与发布工作流',
machine_readiness='guarded_release',
machine_safe_for_codegen=False,
write_path_ready=False,
schema_available=False,
example_available=True,
production_supported=False,
best_for=[
'理解发布门禁与检查项',
'规划发布工作流',
],
blocking_gaps=[
'签名发布、CHANGELOG 验收与发布证据仍未关闭',
],
boundaries=[
'当前发布流程仍包含人工门禁，不能自动执行生产发布。',
],
example_entry_points=[
'release_pack/index.json',
],
),
CapabilitySurface(
surface_id='execution_protocol_pack',
title='执行协议与命令注册中心',
machine_readiness='guarded_execution',
machine_safe_for_codegen=True,
write_path_ready=False,
schema_available=True,
example_available=True,
production_supported=False,
best_for=[
'理解协议执行路由',
'规划自动化工作流',
],
blocking_gaps=[
'执行协议当前仍受门禁保护，不能自动触发生产执行',
],
boundaries=[
'可辅助理解执行路径，但不能直接用于生产自动化。',
],
example_entry_points=[
'execution_pack/index.json',
'execution_pack/command_registry.json',
],
),
]
def print_full_matrix() -> None:
"""打印完整的能力矩阵摘要。"""
for surface in CAPABILITY_MATRIX:
print(surface.summary_line())
def get_surfaces_by_readiness(
readiness: str,
) -> List[CapabilitySurface]:
"""按机器就绪度筛选表面。"""
return [
s for s in CAPABILITY_MATRIX
if s.machine_readiness == readiness
]
def surfaces_safe_for_autogen() -> List[CapabilitySurface]:
"""获取 AI 可安全自动生成代码的表面。"""
return [s for s in CAPABILITY_MATRIX if s.can_autogen]
def surfaces_with_write_path() -> List[CapabilitySurface]:
"""获取写路径可用的表面。"""
return [s for s in CAPABILITY_MATRIX if s.can_write]
def surfaces_production_ready() -> List[CapabilitySurface]:
"""获取生产就绪的表面。"""
return [s for s in CAPABILITY_MATRIX if s.is_production]
3.2 按就绪度分组
# 按 production_supported 分组
PRODUCTION_READY_SURFACES = [
'contract_runtime', # assisted_codegen, write ready
'registry_resolution', # production_reference
'rpc_gateway', # assisted_codegen, write ready
'chain_config_pack', # starter_ready
]
# 有写路径但非生产
WRITE_READY_NON_PRODUCTION = [
'wallet_frontend', # guarded_integration
'dapp_starter_pack', # starter_ready
]
# 仅可读
READ_ONLY_SURFACES = [
'core_contract_reference_pack', # source_backed_reference
'formal_api_schema_pack', # source_backed_reference
'explorer_receipts', # read_only_assist
'contract_template_pack', # starter_ready
'developer_quickstart_pack', # starter_ready
]
# 不可自动执行（需人类门禁）
HUMAN_GATED_SURFACES = [
'agent_query_and_guarded_write', # guarded_write
'sdk_surface', # local_candidate
'public_sandbox_strategy', # fail_closed_reference
'release_pack', # guarded_release
'execution_protocol_pack', # guarded_execution
]
def classify_surface(surface_id: str) -> str:
"""分类表面的当前使用建议。"""
if surface_id in PRODUCTION_READY_SURFACES:
return 'PRODUCTION — 可安全用于生产'
elif surface_id in WRITE_READY_NON_PRODUCTION:
return 'CAUTION — 写可用但非生产，仅限开发/测试'
elif surface_id in READ_ONLY_SURFACES:
return 'REFERENCE — 仅读取，不可写'
elif surface_id in HUMAN_GATED_SURFACES:
return 'HUMAN_GATED — 不可自动执行，需人类审批'
return 'UNKNOWN'
3.3 AI Agent 安全使用建议
from enum import Enum
class AgentAction(Enum):
READ_QUERY = 'read_query'
CODEGEN = 'codegen'
GUARDED_WRITE = 'guarded_write'
AUTONOMOUS_WRITE = 'autonomous_write'
PRODUCTION_DEPLOY = 'production_deploy'
ACTION_RECOMMENDATIONS = {
AgentAction.READ_QUERY: {
'allowed_surfaces': [
'contract_runtime', 'registry_resolution', 'rpc_gateway',
'explorer_receipts', 'chain_config_pack',
],
'caution': 'Query 端点公开可用，但 live/public explorer 尚未全量关闭',
},
AgentAction.CODEGEN: {
'allowed_surfaces': [
'contract_runtime', 'core_contract_reference_pack',
'registry_resolution', 'rpc_gateway',
'formal_api_schema_pack', 'chain_config_pack',
'contract_template_pack', 'dapp_starter_pack',
'developer_quickstart_pack',
],
'caution': (
'AI 可辅助生成代码，但 store/instantiate/migrate 等高风险动作 '
'仍需人工审核与真实部署账户'
),
},
AgentAction.GUARDED_WRITE: {
'allowed_surfaces': [
'wallet_frontend', 'dapp_starter_pack',
],
'caution': (
'写路径可用但非生产环境；'
'所有写操作必须经过 API Key 认证'
),
},
AgentAction.AUTONOMOUS_WRITE: {
'allowed_surfaces': [],
'caution': (
'当前没有任何表面支持完全自动化的生产写操作。'
'agent_query_and_guarded_write 表面明确为 guarded_write，'
'不可 codegen safe'
),
},
AgentAction.PRODUCTION_DEPLOY: {
'allowed_surfaces': [
'contract_runtime', 'registry_resolution',
'rpc_gateway', 'chain_config_pack',
],
'caution': (
'这些表面的核心路径已 not independently verified for production，但完整的生产部署 '
'仍需要真实权限、多签治理与验收流程'
),
},
}
def recommend_surfaces(action: AgentAction) -> List[str]:
"""根据 Agent 动作类型推荐适用的能力表面。"""
rec = ACTION_RECOMMENDATIONS.get(action)
if not rec:
return []
print(f'[建议] {action.value}')
print(f' 适用表面: {", ".join(rec["allowed_surfaces"])}')
print(f' 注意: {rec["caution"]}')
return rec['allowed_surfaces']
4. Stub 端点处理
4.1 Stub 端点检测
Stub 端点返回的 HTTP 响应特征:
- 响应头:
X-MSG-Stub: true - 响应体:
{"message": "stub - not yet implemented"}
import httpx
from typing import Optional, Dict, Any, Union
import logging
logger = logging.getLogger('msg_agent')
class StubResponse:
"""Stub 端点响应封装。"""
def __init__(self, message: str, original_response: Any = None):
self.message = message
self.is_stub = True
self.original_response = original_response
def __repr__(self) -> str:
return f'StubResponse(message="{self.message}")'
def is_stub_response(response: httpx.Response) -> bool:
"""检测响应是否为 Stub。"""
return response.headers.get('X-MSG-Stub', '').lower() == 'true'
class AgentAPIClient:
"""
统一的 Agent API 客户端。
自动检测 Stub 端点并提供优雅降级。
"""
def __init__(
self,
api_key: Optional[str] = None,
base_url: str = MSG_AGENT_BASE,
stub_fallback: Optional[Dict[str, Any]] = None,
):
self.base_url = base_url
self.api_key = api_key
self.stub_fallback = stub_fallback or {}
headers = {}
if api_key:
headers['X-API-Key'] = api_key
self._client = httpx.AsyncClient(
headers=headers, timeout=30.0
)
async def _request(
self,
method: str,
path: str,
**kwargs,
) -> Union[Dict[str, Any], StubResponse]:
"""发送请求并检测 Stub。"""
url = f'{self.base_url}{path}'
try:
resp = await self._client.request(method, url, **kwargs)
if is_stub_response(resp):
data = resp.json()
msg = data.get('message', 'stub - not yet implemented')
logger.warning(f'[STUB] {method} {path}: {msg}')
fallback = self.stub_fallback.get(path)
if fallback:
logger.info(f'使用 fallback 数据: {path}')
return fallback
return StubResponse(msg, original_response=data)
resp.raise_for_status()
return resp.json()
except httpx.HTTPStatusError as e:
if is_stub_response(e.response):
msg = 'stub - not yet implemented'
logger.warning(f'[STUB] {method} {path}: {msg}')
return StubResponse(msg)
raise
async def query_account(
self, address: str
) -> Union[Dict[str, Any], StubResponse]:
return await self._request(
'GET', '/query/account',
params={'address': address},
)
async def query_balance(
self, account: str, asset: str
) -> Union[Dict[str, Any], StubResponse]:
return await self._request(
'GET', '/query/balance',
params={'account': account, 'asset': asset},
)
async def wallet_action(
self, action: str, payload: Dict[str, Any]
) -> Union[Dict[str, Any], StubResponse]:
if not self.api_key:
raise PermissionError('Wallet 操作需要 API Key')
return await self._request(
'POST', f'/wallet/{action}',
json=payload,
)
async def events_history(
self, limit: int = 100
) -> Union[Dict[str, Any], StubResponse]:
return await self._request(
'GET', '/events/history',
params={'limit': limit},
)
async def defi_action(
self, action: str, payload: Dict[str, Any]
) -> Union[Dict[str, Any], StubResponse]:
"""访问 DeFi 端点 — 当前为 Stub。"""
return await self._request(
'POST', f'/defi/{action}',
json=payload,
)
async def close(self) -> None:
await self._client.aclose()
4.2 优雅降级策略
from typing import TypeVar, Callable, Awaitable
import asyncio
T = TypeVar('T')
async def with_stub_fallback(
primary: Callable[[], Awaitable[Union[Dict[str, Any], StubResponse]]],
fallback_data: Dict[str, Any],
endpoint_name: str,
) -> Dict[str, Any]:
"""
通用 Stub 降级包装器。
- 如果返回 StubResponse，使用 fallback_data
- 如果抛出异常，记录并返回 fallback_data
"""
try:
result = await primary()
if isinstance(result, StubResponse):
logger.warning(
f'{endpoint_name} 返回 stub，使用 fallback 数据'
)
return fallback_data
return result
except Exception as e:
logger.error(
f'{endpoint_name} 调用失败: {e}，使用 fallback 数据'
)
return fallback_data
async def compose_agent_workflow(
client: AgentAPIClient,
address: str,
) -> Dict[str, Any]:
"""
组合工作流示例：查询 + 如果非 Stub 则执行操作。
"""
# 第 1 步: 查询余额 (真实可用)
balance = await client.query_balance(address, 'umsg')
if isinstance(balance, StubResponse):
logger.error('余额查询失败 — 工作流中止')
return {'status': 'failed', 'reason': 'balance_query_stub'}
amount = int(balance.get('amount', 0))
if amount < 1_000_000:
return {
'status': 'insufficient_funds',
'balance': amount,
}
# 第 2 步: 尝试 DeFi 操作 (Stub)
defi_result = await client.defi_action('stake', {
'account': address,
'amount': str(amount // 2),
})
if isinstance(defi_result, StubResponse):
logger.warning('DeFi 操作为 stub — 跳过')
return {
'status': 'partial',
'balance': amount,
'defi': 'skipped_stub',
}
4.3 Stub 端点清单与期望行为
STUB_ENDPOINTS = {
'/defi/stake': {
'expected_status': 'stub',
'fallback_behavior': '跳过 DeFi 操作',
'roadmap': '待实现',
},
'/defi/withdraw': {
'expected_status': 'stub',
'fallback_behavior': '跳过提款操作',
'roadmap': '待实现',
},
'/defi/claim': {
'expected_status': 'stub',
'fallback_behavior': '跳过领取操作',
'roadmap': '待实现',
},
'/oracle/query': {
'expected_status': 'stub',
'fallback_behavior': '使用本地缓存的预言机数据',
'roadmap': '开发中',
},
'/oracle/submit': {
'expected_status': 'stub',
'fallback_behavior': '跳过预言机提交',
'roadmap': '开发中',
},
'/payment/create': {
'expected_status': 'stub',
'fallback_behavior': '跳过支付通道创建',
'roadmap': '待实现',
},
'/payment/claim': {
'expected_status': 'stub',
'fallback_behavior': '跳过支付领取',
'roadmap': '待实现',
},
'/bridge/deposit': {
'expected_status': 'stub',
'fallback_behavior': '跳过跨链桥存款',
'roadmap': '待实现',
},
'/bridge/withdraw': {
'expected_status': 'stub',
'fallback_behavior': '跳过跨链桥提款',
'roadmap': '待实现',
},
}
def is_known_stub(path: str) -> bool:
"""判断路径是否为已知 Stub 端点。"""
return path in STUB_ENDPOINTS
def get_stub_fallback_data(path: str) -> Optional[Dict[str, Any]]:
"""获取 Stub 端点的推荐 fallback 数据。"""
fallbacks = {
'/defi/stake': {
'status': 'stub',
'message': 'stub - not yet implemented',
'fallback': True,
},
'/oracle/query': {
'status': 'stub',
'message': 'stub - not yet implemented',
'fallback': True,
'hint': 'use on-chain oracle contract directly',
},
}
return fallbacks.get(path)
5. 写操作门禁
5.1 API Key 要求
写路径（Wallet / Registry register / MPC / Events subscribe）必须
提供 API Key。未提供时将返回 401 或 403。
class AuthError(Exception):
"""认证相关错误。"""
pass
class GuardedWriteError(Exception):
"""受保护写操作错误 — 需人类审批。"""
def __init__(self, gate: str, next_step: str):
self.gate = gate
self.next_step = next_step
super().__init__(
f'需要人类审批门禁: {gate}. 下一步: {next_step}'
)
class APIKeyManager:
"""
API Key 管理器。
真实实现应从安全存储读取密钥。
"""
def __init__(self, key_store: Optional[Dict[str, str]] = None):
self._keys = key_store or {}
def get_key(self, env_name: str = 'MSG_API_KEY') -> str:
"""从环境变量或存储中获取 API Key。"""
import os
key = os.environ.get(env_name)
if key:
return key
key = self._keys.get(env_name)
if key:
return key
raise AuthError(
f'未找到 API Key (环境变量 {env_name})'
)
def validate_key_format(self, key: str) -> bool:
"""验证 API Key 格式。"""
return bool(key) and len(key) >= 16
class GuardedWalletClient:
"""
受保护的 Wallet 客户端。
所有写操作都需要 API Key + 人类审批门禁。
"""
STATUS_STUB = 'stub'
STATUS_PENDING_HUMAN_APPROVAL = 'pending_human_approval'
STATUS_REJECTED = 'rejected'
STATUS_READY_FOR_SIGNING = 'ready_for_signing'
def __init__(
self,
api_key: str,
base_url: str = MSG_AGENT_BASE,
):
self.api_key = api_key
self.base_url = base_url
self._client = httpx.AsyncClient(
headers={'X-API-Key': api_key},
timeout=30.0,
)
async def guarded_action(
self,
action: str,
account: str,
unsigned_payload: Dict[str, Any],
justification: str,
) -> Dict[str, Any]:
"""
执行受保护的钱包操作。
返回状态之一:
- stub: 端点未实现
- pending_human_approval: 等待人类审批
- rejected: 被拒绝
- ready_for_signing: 可签名
参数:
action: 操作类型 (如 prepare_sign_transfer)
account: 账户地址
unsigned_payload: 未签名的交易载荷
justification: 操作理由
"""
resp = await self._client.post(
f'{self.base_url}/wallet/{action}',
json={
'account': account,
'action': action,
'unsigned_payload': unsigned_payload,
'justification': justification,
},
)
resp.raise_for_status()
data = resp.json()
status = data.get('status')
if status == self.STATUS_PENDING_HUMAN_APPROVAL:
raise GuardedWriteError(
gate=data.get('approval_gate', 'unknown'),
next_step=data.get(
'next_step',
'request operator-provided signer or remain in plan mode',
),
)
elif status == self.STATUS_REJECTED:
raise GuardedWriteError(
gate='rejected',
next_step='操作已被拒绝，请检查理由',
)
elif status == self.STATUS_STUB:
logger.warning(f'Wallet/{action} 为 stub')
return data
async def close(self) -> None:
await self._client.aclose()
5.2 人类审批门禁
来自 agent_surface.yaml 的 x-msg-human-approval-gates 字段明确
标识了需要人类介入的门禁点:
HUMAN_APPROVAL_GATES = {
'secret_injection': {
'description': '私钥或助记词的注入需要人类操作员提供签名者',
'affected_endpoints': [
'POST /agent/v1/wallet/sign',
'POST /agent/v1/wallet/transfer',
],
'severity': 'critical',
},
'production_release': {
'description': '生产发布需要多签治理与审批窗口',
'affected_endpoints': [
'POST /agent/v1/wallet/broadcast',
'POST /agent/v1/registry/register',
],
'severity': 'high',
},
}
def check_approval_gates(
endpoint: str, payload: Dict[str, Any]
) -> List[str]:
"""检查端点是否触发人类审批门禁。"""
triggered = []
for gate_name, gate in HUMAN_APPROVAL_GATES.items():
if endpoint in gate['affected_endpoints']:
triggered.append(gate_name)
return triggered
5.3 高风险动作边界
HIGH_RISK_ACTIONS = {
'store_code': {
'risk': '上传合约字节码',
'required_gates': ['governance_approval', 'security_review'],
'surface': 'contract_runtime',
},
'instantiate': {
'risk': '实例化合约',
'required_gates': ['governance_approval'],
'surface': 'contract_runtime',
},
'migrate': {
'risk': '迁移合约',
'required_gates': ['governance_approval', 'security_review'],
'surface': 'contract_runtime',
},
'treasury_withdraw': {
'risk': '金库提款',
'required_gates': ['multi_sig', 'timelock'],
'surface': 'N/A (不在当前 Agent API 中)',
},
'validator_stake': {
'risk': '验证者质押',
'required_gates': ['user_confirmation'],
'surface': 'N/A (通过原生 Cosmos SDK)',
},
}
def is_high_risk(action: str) -> bool:
return action in HIGH_RISK_ACTIONS
def required_gates_for(action: str) -> List[str]:
return HIGH_RISK_ACTIONS.get(action, {}).get(
'required_gates', []
)
6. 当前缺口与路线图
6.1 当前已识别缺口
基于 developer_capability_matrix.json 各表面的 blocking_gaps
和白皮书边界条款，以下缺口已在官方文档中明确标识:
KNOWN_GAPS = {
'governance_native_api': {
'severity': 'high',
'description': 'Governance native API 缺失 — 当前 Agent API 不含治理提案、'
'投票、参数修改等原生产点',
'affected_use_cases': [
'DAO 操作自动化',
'治理参数查询与修改',
'提案生命周期管理',
],
'workaround': '通过 CosmWasm 合约接口间接操作治理模块',
'status': 'not_implemented',
},
'validator_lifecycle_api': {
'severity': 'high',
'description': 'Validator lifecycle API 缺失 — Agent API 不含验证者注册、'
'质押、解质押、惩罚查询等端点',
'affected_use_cases': [
'验证者操作自动化',
'质押状态监控',
'DAR 评分查询',
],
'workaround': '通过原生 RPC 接口操作 (非 Agent API)',
'status': 'not_implemented',
},
'contract_execute_unified_api': {
'severity': 'medium',
'description': '缺少统一的合约执行端点 — 当前没有通用 contract/execute 端点',
'affected_use_cases': [
'AI Agent 自动调用任意合约方法',
'跨合约组合操作',
],
'workaround': '通过 JSON-RPC 或 CosmJS 直接发送 execute 消息',
'status': 'not_implemented',
},
'treasury_independent_entry': {
'severity': 'medium',
'description': 'Treasury（金库）没有独立的 Agent 入口 — 金库多签阈值操作'
'需要通过注册中心规范键间接操作',
'affected_use_cases': [
'金库余额自动检查',
'多签提款提案',
],
'workaround': '通过 Registry 解析金库合约地址后直接调用',
'status': 'partial',
},
'signed_sdk_release': {
'severity': 'medium',
'description': 'SDK 尚未形成 signed/public release，当前为 alpha/local candidate',
'affected_use_cases': [
'使用 SDK 构建生产应用',
'依赖稳定公共 API',
],
'workaround': '直接使用 REST/JSON-RPC 端点',
'status': 'alpha',
},
'public_sandbox': {
'severity': 'medium',
'description': '当前为 fail-closed 策略 — 无公开 testnet/faucet',
'affected_use_cases': [
'无需本地节点的快速开发测试',
'CI/CD 集成测试',
],
'workaround': '运行本地开发节点或使用 localhost starter',
'status': 'fail_closed',
},
'live_public_explorer': {
'severity': 'low',
'description': 'live/public Explorer 仍未全量关闭，'
'索引与分页的生产 SLO/保留策略仍需独立交付',
'affected_use_cases': [
'开发参考级区块链浏览器查询',
'高可用事件索引',
],
'workaround': '使用本地节点 RPC 查询',
'status': 'partial',
},
}
def print_gaps_by_severity() -> None:
"""按严重程度打印已知缺口。"""
for sev in ['high', 'medium', 'low']:
print(f'\n=== {sev.upper()} 严重性 ===')
for gap_id, gap in KNOWN_GAPS.items():
if gap['severity'] == sev:
print(f' {gap_id}: {gap["description"]}')
print(f' 变通方案: {gap["workaround"]}')
6.2 推荐开发顺序
DEVELOPMENT_PRIORITIES = {
'P0 — 立即 (核心缺失)': [
{
'id': 'governance_native_api',
'rationale': '治理是 MSG 链的核心价值主张，Agent 无法操作治理是最大缺口',
'estimated_effort': 'large',
'dependencies': [],
},
{
'id': 'contract_execute_unified_api',
'rationale': '统一合约执行端点是 Agent 执行链上操作的基本能力',
'estimated_effort': 'medium',
'dependencies': [
'formal_api_schema_pack',
],
},
],
'P1 — 短期 (提升实用性)': [
{
'id': 'validator_lifecycle_api',
'rationale': '验证者操作是链上治理的重要环节',
'estimated_effort': 'medium',
'dependencies': [
'governance_native_api',
],
},
{
'id': 'treasury_independent_entry',
'rationale': '金库是 DAO 的核心资产，独立入口可大幅提升 Agent 自治能力',
'estimated_effort': 'small',
'dependencies': [
'registry_resolution',
],
},
{
'id': 'public_sandbox',
'rationale': '公开 sandbox 可降低开发者接入门槛',
'estimated_effort': 'medium',
'dependencies': [
'chain_config_pack',
],
},
],
'P2 — 中期 (生态建设)': [
{
'id': 'signed_sdk_release',
'rationale': '稳定 SDK 是生态建设的基础',
'estimated_effort': 'large',
'dependencies': [
'formal_api_schema_pack',
'sdk_surface',
],
},
{
'id': 'live_public_explorer',
'rationale': '开发参考级 Explorer 提升生态透明度',
'estimated_effort': 'medium',
'dependencies': [
'rpc_gateway',
],
},
{
'id': 'defi_oracle_payment_bridge',
'rationale': '将 Stub 端点逐一实现为可用端点',
'estimated_effort': 'xl',
'dependencies': [
'contract_runtime',
'formal_api_schema_pack',
],
},
],
}
def recommend_next_steps() -> None:
"""打印推荐的下一步开发计划。"""
for priority, items in DEVELOPMENT_PRIORITIES.items():
print(f'\n## {priority}')
for item in items:
print(f' - {item["id"]}: {item["rationale"]}')
print(f' 依赖: {item["dependencies"] or "无"}')
6.3 各表面状态汇总
SURFACE_STATUS_SUMMARY = {
'contract_runtime': {
'status': 'implemented',
'production': True,
'recommendation': '可安全用于开发参考级合约开发与 AI codegen',
},
'core_contract_reference_pack': {
'status': 'source_backed_reference',
'production': False,
'recommendation': '适合作为参考，不可用于生产写操作',
},
'registry_resolution': {
'status': 'implemented',
'production': True,
'recommendation': '可安全用于生产寻址',
},
'rpc_gateway': {
'status': 'implemented',
'production': True,
'recommendation': '可安全用于生产查询与广播',
},
'formal_api_schema_pack': {
'status': 'partial',
'production': False,
'recommendation': '适合阅读，非生产最终契约',
},
'wallet_frontend': {
'status': 'partial',
'production': False,
'recommendation': '写路径可用但非生产，用于开发测试',
},
'explorer_receipts': {
'status': 'partial',
'production': False,
'recommendation': '只读辅助，适合调试与回证',
},
'agent_query_and_guarded_write': {
'status': 'partial',
'production': False,
'recommendation': '不可自动执行 — 需人类审批',
},
'sdk_surface': {
'status': 'partial',
'production': False,
'recommendation': 'alpha 级，不可用于生产',
},
'chain_config_pack': {
'status': 'implemented',
'production': True,
'recommendation': '可安全用于生产链配置注入',
},
'public_sandbox_strategy': {
'status': 'partial',
'production': False,
'recommendation': 'fail-closed — 走 local development path',
},
'contract_template_pack': {
'status': 'starter_ready',
'production': False,
'recommendation': '可安全用于 AI codegen，非开发参考级模板',
},
'dapp_starter_pack': {
'status': 'starter_ready',
'production': False,
'recommendation': '可安全用于快速搭建原型',
},
'developer_quickstart_pack': {
'status': 'starter_ready',
'production': False,
'recommendation': '用于引导开发流程',
},
'release_pack': {
'status': 'guarded_release',
'production': False,
'recommendation': '含人工门禁，不可自动发布',
},
'execution_protocol_pack': {
'status': 'guarded_execution',
'production': False,
'recommendation': '可辅助理解执行路径，不可自动化',
},
}
7. 附录: 快速参考
7.1 端点速查表
| 分组 | 端点 | 方法 | 认证 | 状态 |
|---|---|---|---|---|
| Query | /agent/v1/query/account |
GET | 公开 | partial |
| Query | /agent/v1/query/balance |
GET | 公开 | partial |
| Query | /agent/v1/query/tx |
GET | 公开 | partial |
| Query | /agent/v1/query/block |
GET | 公开 | partial |
| Query | /agent/v1/query/batch |
POST | 公开 | partial |
| Wallet | /agent/v1/wallet/transfer |
POST | API Key | partial |
| Wallet | /agent/v1/wallet/sign |
POST | API Key | partial |
| Wallet | /agent/v1/wallet/broadcast |
POST | API Key | partial |
| Events | /agent/v1/events/history |
GET | 公开 | partial |
| Events | /agent/v1/events/subscribe |
POST | API Key | partial |
| Registry | /agent/v1/registry/resolve |
GET | 公开 | partial |
| Registry | /agent/v1/registry/discover |
GET | 公开 | partial |
| Registry | /agent/v1/registry/register |
POST | API Key | partial |
| MPC | /agent/v1/mpc/keygen |
POST | API Key | partial |
| MPC | /agent/v1/mpc/sign |
POST | API Key | partial |
| MPC | /agent/v1/mpc/aggregate |
POST | API Key | partial |
| DeFi | /agent/v1/defi/* |
* | N/A | stub |
| Oracle | /agent/v1/oracle/* |
* | N/A | stub |
| Payment | /agent/v1/payment/* |
* | N/A | stub |
| Bridge | /agent/v1/bridge/* |
* | N/A | stub |
7.2 关键文件索引
KEY_DOCUMENTS = {
'Agent API OpenAPI (YAML)': {
'url': 'https://msgchain.org/whitepaper/api_specs/openapi/agent_surface.yaml',
'description': 'OpenAPI 3.1 规范 — Agent 查询面与受保护写路径',
},
'Agent API 能力表面模块': {
'url': 'https://msgchain.org/whitepaper/modules/agent_api_surface.html',
'description': '白皮书模块页面，含流程图与能力分析',
},
'Agent API 机器导出': {
'url': 'https://msgchain.org/whitepaper/module_exports/agent_api_surface.json',
'description': '机器可读的模块导出（标签、外链、边界条款）',
},
'能力成熟度矩阵': {
'url': 'https://msgchain.org/whitepaper/developer_capability_matrix.json',
'description': '16 个表面的机器就绪度评级',
},
'开发者机器入口': {
'url': 'https://msgchain.org/whitepaper/developer_entry.json',
'description': '推荐消费顺序、入口点与边界条款',
},
'Agent 机器入口': {
'url': 'https://msgchain.org/whitepaper/agent_entry.json',
'description': '爬虫协议、入口点索引与领域映射',
},
'正式 API/Schema 契约索引': {
'url': 'https://msgchain.org/whitepaper/api_specs/formal_contracts.json',
'description': 'RPC 方法、OpenAPI 摘要、错误代码的契约索引',
},
'RPC 方法清单': {
'url': 'https://msgchain.org/whitepaper/api_specs/rpc_methods.json',
'description': '完整 RPC 方法清单与签名',
},
'错误代码': {
'url': 'https://msgchain.org/whitepaper/api_specs/error_codes.json',
'description': '标准错误代码与含义',
},
'Public Query OpenAPI': {
'url': 'https://msgchain.org/whitepaper/api_specs/openapi/public_query.yaml',
'description': '公开查询端点的 OpenAPI 规范',
},
'合约表面 OpenAPI': {
'url': 'https://msgchain.org/whitepaper/api_specs/openapi/contract_surface.yaml',
'description': '合约交互表面的 OpenAPI 规范',
},
'沙箱策略': {
'url': 'https://msgchain.org/whitepaper/chain_config/developer_sandbox_strategy.json',
'description': '当前的 fail-closed 沙箱策略',
},
'AI Agent 启动提示词': {
'url': 'https://msgchain.org/whitepaper/integration_examples/external_ai_agent_bootstrap_prompt.json',
'description': '外部 AI Agent 的引导提示词模板',
},
}
7.3 边界条款 — 不可越级宣称
白皮书系统明确标注了以下边界条款，AI Agent 必须遵守:
BOUNDARY_CLAUSES = [
# 来自 agent_api_surface module_export
'不能把 OpenAPI 表面等同成全部能力已实现',
'OpenAPI 中出现的全部端点都已可生产使用 → FALSE',
'DeFi / Payment / Bridge 等已形成真实闭环 → FALSE',
'Agent API 已完全覆盖人类可执行的链上动作 → FALSE',
'接口表面真实存在，能力成熟度分层明显',
'部分分类仍有 Stub，不能把接口表面当成交付完成',
'高风险接口不能裸放权',
# 来自 developer_capability_matrix.json
'Stub responses and guarded writes must never be treated '
'as autonomous production completion',
# 来自 agent_surface.yaml
'This path must remain guarded and cannot be interpreted '
'as autonomous production signing',
# 通用原则
'AI 可辅助生成合约代码，但仍需结合业务规则、'
'人类审核与真实部署账户完成落地',
'涉及私钥、部署权限、生产域名、资金操作、'
'DAO/timelock/threshold 的动作，必须保留人类确认与审批门禁',
]
本文档结束 — 生成自 MSG Chain 白皮书实时数据。
最后同步: 2026-07
数据来源: https://msgchain.org/whitepaper/所有代码示例均使用
msg前缀命名空间，以反映 MSG Chain 的命名惯例。