Agent Governance Toolkit 协议桥接实战:用信任门控打通 MCP、A2A 与 IATP 异构智能体网络
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
本教程(对应仓库 docs/tutorials/16-protocol-bridges.md)讲解 Agent Governance Toolkit 中三大协议桥接组件——MCP(Model Context Protocol)、A2A(Agent-to-Agent)与 IATP(Inter-Agent Trust Protocol)——的集成与使用。你将在 30 分钟内学会:用
TrustProxy为任意 MCP 服务器套上信任门控中间件、用TaskEnvelope与TrustGate治理 A2A 任务委托、用 Ed25519 握手与签名 Agent Card 建立跨协议信任,并通过ProtocolBridge与agentmesh proxyCLI 打通异构智能体之间的消息翻译、审计与编排。阅读前提:Python 3.11+,安装pip install agentmesh-platform a2a-agentmesh mcp-trust-proxy。
1. 为什么需要协议桥接
AI Agent 之间说着不同的"方言":一个 MCP 编码助手、一个 A2A 任务智能体、一个 IATP 治理节点可能需要在同一业务流里协作,但它们之间没有共享的线格式(wire format)、信任模型或发现机制。协议桥接(Protocol Bridge)解决的正是这个问题——翻译消息、以信任为门槛放行通信、并完整保留审计轨迹。
Agent Governance Toolkit 内置三套桥接组件,对应仓库agent-governance-python/agentmesh-integrations/下的独立包:
| 协议 | 用途 | 关键包 |
|---|---|---|
| MCP(Model Context Protocol) | 为 AI Agent 提供受信任门控的工具暴露 | agentmesh.cli.proxy、mcp_trust_proxy |
| A2A(Agent-to-Agent) | 任务委托与智能体发现 | a2a_agentmesh、agent_os.integrations.a2a_adapter |
| IATP(Inter-Agent Trust Protocol) | 认证(Attestation)、握手与声誉追踪 | agentmesh.trust.handshake、iatp.proto |
本文将按如下路线推进:先跑通 MCP 信任代理的最小示例,再逐一深入 MCP、A2A、IATP 三个桥接层,随后介绍加密签名的 Agent Card、跨协议统一信任门控、多协议编排,最后落到agentmesh proxy命令行工具的治理代理。
2. 快速开始:5 行代码为 MCP 服务器套上信任门控
任何一个 MCP 服务器都可以用TrustProxy包一层治理,核心只有两件事:声明按工具(per-tool)策略,然后在转发工具调用前调用authorize()。
from mcp_trust_proxy import TrustProxy, ToolPolicy # 创建带按工具策略的信任代理 proxy = TrustProxy( default_min_trust=300, tool_policies={ "file_write": ToolPolicy(min_trust=800, required_capabilities=["fs_write"]), "shell_exec": ToolPolicy(blocked=True), }, ) # Agent 请求工具访问 result = proxy.authorize( agent_did="did:mesh:agent-1", agent_trust_score=600, agent_capabilities=["fs_read", "search"], tool_name="file_read", ) print(result.allowed) # True — 600 ≥ 300 默认阈值 print(result.reason) # "Authorized" print(result.trust_score) # 600这段代码对应 mcp_trust_proxy/proxy.py 中的核心实现:ToolPolicy数据类定义了min_trust、required_capabilities、blocked、max_calls_per_minute与description五个字段(max_calls_per_minute=0表示不限流),而authorize()在放行前按固定顺序执行 6 项检查(详见下一节)。值得注意的是,该文件头部注释明确说明它不依赖 MCP SDK——它只操作工具名与智能体元数据,因此可以作为任何 MCP 服务器前的中间件。
3. MCP 协议桥:信任门控的工具暴露
MCP 桥接层拦截 MCP 客户端(如 Claude Desktop、VS Code)与 MCP 工具服务器之间的 JSON-RPC 2.0 消息,对每一次tools/call请求强制执行策略。
┌──────────────┐ ┌─────────────────┐ ┌──────────────┐ │ MCP Client │ ──► │ AgentMesh │ ──► │ MCP Server │ │ (Claude, │ ◄── │ Trust Proxy │ ◄── │ (filesystem,│ │ VS Code) │ │ │ │ database) │ └──────────────┘ └─────────────────┘ └──────────────┘ ● Policy check ● Rate limiting ● Trust score gate ● Audit logging3.1 TrustProxy——适用于任意 MCP 服务器的中间件
TrustProxy以中间件形态工作,require_did=True时强制要求调用方出示 DID,blocked_dids支持拉黑已知受损智能体;set_tool_policy()可以随时为某个工具动态调整策略:
from mcp_trust_proxy import TrustProxy, ToolPolicy proxy = TrustProxy( default_min_trust=100, require_did=True, blocked_dids=["did:mesh:compromised-agent"], ) proxy.set_tool_policy("database_query", ToolPolicy( min_trust=500, required_capabilities=["db_read"], max_calls_per_minute=30, )) proxy.set_tool_policy("shell_exec", ToolPolicy(blocked=True))3.2 授权管线:6 项按序检查
TrustProxy.authorize()按顺序执行以下检查,任一失败立即返回AuthResult(allowed=False, reason=...):
| # | 检查项 | 失败示例 |
|---|---|---|
| 1 | Agent DID 存在 | "Agent DID is required" |
| 2 | DID 未被拉黑 | "Agent {did} is blocked" |
| 3 | 工具未被禁用 | "Tool '{name}' is blocked by policy" |
| 4 | 信任分 ≥ 阈值 | "Trust score {n} below minimum {m}" |
| 5 | 具备所需能力 | "Missing capabilities: [...]" |
| 6 | 未超速率限制 | "Rate limit exceeded ({n}/min)" |
在源码 proxy.py 中还可以看到第 5 项与第 6 项之间隐藏的5.5 项检查:工具参数注入扫描。TrustProxy内置了四类提示注入正则(INJECTION_PATTERNS):direct_override(如 "ignore all previous instructions")、roleplay_jailbreak(伪装 system/admin/root)、system_prompt_exfiltration(诱导输出系统提示词)、delimiter_attack(###、---END---、[SYSTEM]、<|im_start|>等分隔符)。_scan_tool_args()会递归遍历工具参数的字符串值做匹配;若扫描本身抛异常,则按 fail-closed 原则直接拒绝。
典型的桥接层集成方式是在 MCP 服务器返回错误之前拦截:
result = proxy.authorize( agent_did="did:mesh:data-agent", agent_trust_score=750, agent_capabilities=["db_read", "search"], tool_name="database_query", ) if not result.allowed: return {"jsonrpc": "2.0", "error": {"code": -32001, "message": result.reason}}3.3 审计轨迹
每一次授权决策都会被追加进_audit_log,可通过get_audit_log()读取;get_stats()返回聚合统计:
for entry in proxy.get_audit_log(): print(f"{entry.tool_name}: {'✅' if entry.allowed else '❌'} " f"(agent={entry.agent_did}, trust={entry.trust_score})") stats = proxy.get_stats() # {"total_requests": 142, "allowed": 130, "denied": 12, ...}get_stats()实际返回的字段还包括configured_tools(已配置策略的工具数)与blocked_agents(被拉黑智能体数);block_agent()/unblock_agent()可运行时维护黑名单,clear_rate_limits()可重置限流计数。源码注释同时提醒:TrustProxy不是线程安全的,并发场景请用threading.Lock包住authorize()。
3.4 MCP 治理策略(YAML)
MCPProxyCLI 加载 YAML 治理策略,用rules定义动作(allow / require_approval / deny)、用trust块配置信任分阈值与拒绝扣分、用audit块开启审计:
policies: - id: mcp-claude-desktop rules: - id: allow-reads action: allow conditions: ["tool in ['read_file', 'search_files']"] - id: approve-writes action: require_approval conditions: ["tool in ['write_file', 'execute_command']"] - id: block-destructive action: deny conditions: ["tool in ['delete_file', 'shell_exec']"] trust: {min_score: 600, score_penalty_on_deny: 10} audit: {enabled: true, format: cloudevents}这种action + conditions的声明式策略与 Tutorial 01 — Policy Engine 介绍的 YAML 策略引擎一脉相承,也是第 9 节 CLI 代理的底层判定依据。
4. A2A 协议桥:任务委托与智能体发现
A2A(Agent-to-Agent)是跨框架智能体互操作的开源标准。Toolkit 提供两层集成:
a2a_agentmesh—— 独立桥接包,含AgentCard、TaskEnvelope、TrustGate三个组件(位于 a2a-protocol/a2a_agentmesh/);agent_os.integrations.a2a_adapter—— Agent-OS 内核级治理适配器(位于 agent_os/integrations/a2a_adapter.py)。
4.1 Agent Card——通过/.well-known/agent.json发现
Agent Card 是 A2A 标准的"智能体名片",发布在/.well-known/agent.json供发现。AgentMesh 通过x-agentmesh-*扩展字段注入信任元数据:
from a2a_agentmesh import AgentCard card = AgentCard.from_identity( did="did:mesh:translator-01", name="TranslationAgent", description="Translates text between 40+ languages", capabilities=["translate", "detect_language", "summarize"], trust_score=800, organization="Linguatech", ) print(card.to_json())输出包含标准 A2A 字段与 AgentMesh 扩展:
{ "name": "TranslationAgent", "skills": [{"id": "translate", "name": "Translate"}, ...], "authentication": {"schemes": ["iatp"]}, "x-agentmesh-did": "did:mesh:translator-01", "x-agentmesh-trust-score": 800, "x-agentmesh-protocols": ["a2a/1.0", "iatp/1.0"] }card.has_skill("translate") # True card.skill_ids() # ["translate", "detect_language", "summarize"]从 agent_card.py 可以看到映射细节:capabilities列表中的每一项会被映射成一个 A2Askills[]条目(每个能力一个AgentSkill,默认输入/输出模式为text/plain),同时x-agentmesh-*扩展还包含x-agentmesh-organization与x-agentmesh-public-key-fingerprint等字段;from_dict()则支持从线上抓取的agent.json反序列化。
4.2 Task Envelope——带信任验证的任务生命周期
任务遵循 A2A 状态机:submitted → working → complete/failed/canceled。AgentMesh 将信任证明嵌入x-agentmesh-trust字段:
from a2a_agentmesh import TaskEnvelope, TaskState task = TaskEnvelope.create( skill_id="translate", source_did="did:mesh:orchestrator-01", target_did="did:mesh:translator-01", source_trust_score=750, input_text="Translate 'governance matters' to Spanish", ) task.start() # → WORKING task.complete("La gobernanza importa") # → COMPLETE print(task.is_terminal) # True # Invalid transitions raise ValueError序列化为 JSON-RPC 兼容格式传输:
wire_format = task.to_dict() # Includes: id, status.state, skill_id, messages[], x-agentmesh-trust{} received = TaskEnvelope.from_dict(wire_format)源码层面 task.py 用_TRANSITIONS表强制执行合法迁移:SUBMITTED可转到WORKING/CANCELED/FAILED,WORKING可转到COMPLETE/FAILED/CANCELED,终态(COMPLETE/FAILED/CANCELED)不可再迁移,非法迁移抛出ValueError并列出允许的目标状态;to_dict()序列化时仅在存在相应字段时才写入x-agentmesh-trust子键(source_did/target_did/source_trust_score)。
4.3 Trust Gate——A2A 入站任务的策略执行
TrustGate对入站 A2A 任务按策略评估,支持最小信任分、按来源的限流、DID 黑/白名单,以及按技能的信任分覆盖:
from a2a_agentmesh import TrustGate, TrustPolicy, TaskEnvelope gate = TrustGate(TrustPolicy( min_trust_score=500, max_requests_per_minute=30, blocked_dids=["did:mesh:banned-agent"], skill_trust_overrides={"admin_action": 900, "search": 200}, require_did=True, )) # 评估入站任务 task = TaskEnvelope.create( skill_id="translate", source_did="did:mesh:caller", source_trust_score=600, input_text="Hello", ) result = gate.evaluate(task) print(result.allowed, result.reason) # True, "Trust verified" # 自动失败被拒任务 low_trust = TaskEnvelope.create( skill_id="admin_action", source_did="did:mesh:newbie", source_trust_score=300, input_text="Delete all users", ) result = gate.evaluate_and_gate(low_trust) # low_trust.state == FAILED, low_trust.error contains denial reason实现细节见 trust_gate.py:evaluate()依次执行 5 项检查——DID 存在(require_did)、DID 黑名单、DID 白名单(allowed_dids非空时启用)、信任分(优先取skill_trust_overrides[skill_id],否则取min_trust_score)、每分钟限流。evaluate_and_gate()在拒绝时自动把任务转入FAILED状态并写入error;get_stats()可查看总评估数、放行/拒绝数与受追踪的智能体数。
4.4 A2A 治理适配器(Agent-OS 内核级)
需要内核级治理时,A2AGovernanceAdapter在信任门控之上叠加内容过滤。当前仓库中的实现(agent_os/integrations/a2a_adapter.py)通过 AGT 5.0 ACS 原生运行时(AdapterRuntime)的input干预点对任务文本做内容检查,并支持可选的conversation_guardian会话守护:
from agent_os.integrations.a2a_adapter import A2AGovernanceAdapter adapter = A2AGovernanceAdapter( allowed_skills=["search", "translate"], blocked_patterns=["DROP TABLE", "rm -rf"], min_trust_score=300, ) result = adapter.evaluate_task({ "skill_id": "search", "x-agentmesh-trust": {"source_did": "did:mesh:a", "source_trust_score": 500}, "messages": [{"role": "user", "parts": [{"text": "Find weather"}]}], }) print(result.allowed) # True # 被拦截的内容 result = adapter.evaluate_task({ "skill_id": "search", "x-agentmesh-trust": {"source_did": "did:mesh:b", "source_trust_score": 900}, "messages": [{"role": "user", "parts": [{"text": "DROP TABLE users;"}]}], }) print(result.allowed, result.reason) # False, "Content matches blocked pattern..."该适配器的evaluate_task()依次执行:信任元数据检查、技能黑名单、技能白名单、信任分阈值、内容检查(经 ACS 输入干预点)、会话守护分析与限流。A2AEvaluation还会携带transform_value(AGT-DELTA D1.1 重写结果,宿主可据此改写外发负载)与conversation_alert,评估结果可由get_evaluations()与get_stats()回溯。
5. IATP——智能体间信任协议
IATP 是 Toolkit 的原生信任协议,提供密码学握手、代码库认证(codebase attestation)与声誉追踪;它以Protocol Buffers定义,具备跨语言互操作性(Python / Node.js / Go / Rust / Java,经 gRPC 通信)。
5.1 信任握手(Ed25519 挑战/响应)
TrustHandshake实现四步验证:挑战(随机 nonce)→响应(Ed25519 签名)→验证(注册表核查)→结果(信任等级 + 能力集)。
from agentmesh.identity import AgentIdentity, IdentityRegistry from agentmesh.trust.handshake import TrustHandshake registry = IdentityRegistry() agent_a = AgentIdentity.create(name="Orchestrator", sponsor="ops@co.com", capabilities=["orchestrate"]) agent_b = AgentIdentity.create(name="DataAgent", sponsor="data@co.com", capabilities=["read:data", "query:db"]) registry.register(agent_a) registry.register(agent_b) handshake = TrustHandshake( agent_did=str(agent_a.did), identity=agent_a, registry=registry, ) result = await handshake.initiate( peer_did=str(agent_b.did), protocol="iatp", required_trust_score=500, required_capabilities=["read:data"], ) if result.verified: print(result.trust_level) # "trusted" print(result.trust_score) # 800 print(result.capabilities) # ["read:data", "query:db"] print(f"{result.latency_ms}ms")信任等级由分数区间映射:
Ring 0 (≥900) ████████████████████████████████████████ Verified Partner Ring 1 (≥700) ██████████████████████████████ Trusted Ring 2 (≥400) ████████████████████ Standard Ring 3 (<400) ██████████ Untrusted 0 200 400 600 800 10005.2 握手缓存
验证过的对端会被缓存(默认 15 分钟),避免重复执行密码学运算。需要强制重新验证时调用handshake.clear_cache();超出超时时间的握手抛出HandshakeTimeoutError。
5.3 IATP Protocol Buffer 定义
IATP 的 proto 定义位于 agent-governance-python/agent-os/modules/iatp/proto/iatp.proto(package iatp.v1)。关键消息:
| 消息 | 用途 |
|---|---|
CapabilityManifest | Agent ID、信任等级、能力清单、隐私契约 |
AttestationRecord | 代码库哈希、配置哈希、签名、有效期 |
HandshakeRequest/Response | Manifest + 认证 + nonce / 会话令牌 |
ActionRequest/Response | 动作名、参数、撤销信息 |
ReputationEvent | 事件类型、严重度、分数增量 |
proto 中还定义了三个 gRPC 服务:TrustProtocol(Handshake、ExecuteAction、UndoAction、StreamActions、HealthCheck)、AttestationService(RequestAttestation、VerifyAttestation)与ReputationService(GetReputation、ReportEvent、StreamReputationUpdates)。配套的PrivacyContract与TracingContext分别承载数据处置策略与分布式追踪上下文(后者与 Tutorial 13 — Observability and Tracing 呼应)。
构建各语言桩代码的命令(proto 头部注释中给出):
# Python python -m grpc_tools.protoc -I. --python_out=. --grpc_python_out=. iatp.proto # Node.js grpc_tools_node_protoc --js_out=. --grpc_out=. iatp.proto # Go protoc --go_out=. --go-grpc_out=. iatp.proto5.4 认证与隐私
AttestationRecord将智能体的代码库绑定到密码学哈希(codebase_hash与config_hash均为 SHA-256,另含 Control Plane 签名signature、signing_key_id、创建与过期时间戳),防止部署后代码被篡改。IATP 握手还包含PrivacyContract,明确数据处置要求:保留策略(RETENTION_POLICY_EPHEMERAL/TEMPORARY/PERMANENT)、存储位置(storage_location)、人工复核许可(human_review)与传输/静态加密要求(encryption_at_rest/encryption_in_transit)。隐私条款不达标的握手可以被对端直接拒绝。
6. 可信 Agent Card:加密签名的智能体元数据
TrustedAgentCard在 A2A Agent Card 之上叠加 Ed25519 密码学签名,防止身份冒充:
from agentmesh.identity import AgentIdentity from agentmesh.trust.cards import TrustedAgentCard, CardRegistry identity = AgentIdentity.create( name="AnalyticsAgent", sponsor="data-team@company.com", capabilities=["analyze", "report", "visualize"], ) # 从身份创建签名卡片 card = TrustedAgentCard.from_identity(identity) print(card.agent_did) # "did:mesh:..." print(card.card_signature) # Base64 Ed25519 signature # 验证——任何持有公钥的一方都能校验 assert card.verify_signature() # 篡改 → 验证失败 card.name = "MaliciousAgent" assert not card.verify_signature()CardRegistry存储已验证卡片,提供基于能力的发现与吊销列表集成:
registry = CardRegistry(cache_ttl_seconds=900) registry.register(card) # Verifies signature first translators = registry.find_by_capability("translate") # Capability discovery is_valid = registry.is_verified("did:mesh:agent-1") # Cached + revocation check7. 信任门控:贯穿所有协议桥的核心原则
在任何跨协议通信发生之前,发起方的信任分必须与阈值比对。这是连接所有协议桥的核心原则。
7.1 TrustBridge——对端验证与 HMAC 缓存
TrustBridge用 HMAC 完整性保护来管理对端信任:
from agentmesh.trust.bridge import TrustBridge bridge = TrustBridge( agent_did="did:mesh:my-agent", default_trust_threshold=700, identity=my_identity, registry=identity_registry, ) # 验证对端(IATP 握手),结果被 HMAC 缓存 result = await bridge.verify_peer( peer_did="did:mesh:remote-agent", protocol="iatp", required_trust_score=600, required_capabilities=["read:data"], ) # 快速缓存检查(校验存储记录的 HMAC 完整性) is_ok = await bridge.is_peer_trusted("did:mesh:remote-agent", required_score=600) trusted = bridge.get_trusted_peers(min_score=800) await bridge.revoke_peer_trust("did:mesh:compromised", reason="anomaly detected")7.2 按协议区分信任阈值
不同协议可执行不同的信任阈值,形成从低风险读到特权操作的分级门控:
| 协议 | 典型阈值 | 使用场景 |
|---|---|---|
| MCP(读工具) | 300 | 低风险数据访问 |
| MCP(写工具) | 800 | 文件系统修改 |
| A2A(任务委托) | 500 | 标准智能体间任务 |
| A2A(管理动作) | 900 | 特权操作 |
| IATP(握手) | 700 | 对端验证默认值 |
8. 多协议编排:A2A ↔ MCP ↔ IATP 的自动翻译与信任验证
ProtocolBridge让智能体跨协议通信,自动完成消息翻译与信任验证:
from agentmesh.trust.bridge import ProtocolBridge bridge = ProtocolBridge( agent_did="did:mesh:orchestrator", identity=orchestrator_identity, registry=identity_registry, ) print(bridge.supported_protocols) # ["a2a", "mcp", "iatp", "acp"]8.1 跨协议消息翻译
桥接层在 A2A 与 MCP 之间翻译消息:
# A2A → MCP: task_type → params.name, parameters → params.arguments a2a_message = {"task_type": "file_analysis", "parameters": {"path": "/data/report.csv"}} result = await bridge.send_message( peer_did="did:mesh:mcp-agent", message=a2a_message, source_protocol="a2a", target_protocol="mcp", ) # Translated to: {"method": "tools/call", "params": {"name": "file_analysis", ...}} # MCP → A2A: params.name → task_type, params.arguments → parameters # IATP → *: IATP wraps any protocol (passthrough)8.2 验证页脚
桥接层可附加验证页脚,证明治理已被应用:
verified = bridge.add_verification_footer( content="Analysis complete.", trust_score=850, agent_did="did:mesh:agent", metadata={"policy": "strict"}, ) # Appends: > 🔒 Verified by AgentMesh (Trust Score: 850/1000)8.3 多智能体工作流:三种协议协同
将三者组合成一条完整链路——编排器经 A2A 委托数据智能体,数据智能体再经 MCP 调用数据库工具:
from agentmesh.identity import AgentIdentity, IdentityRegistry from a2a_agentmesh import TaskEnvelope, TrustGate, TrustPolicy from mcp_trust_proxy import TrustProxy, ToolPolicy # 身份 registry = IdentityRegistry() orchestrator = AgentIdentity.create(name="Orchestrator", sponsor="ops@co.com", capabilities=["orchestrate"]) data_agent = AgentIdentity.create(name="DataAgent", sponsor="data@co.com", capabilities=["query:db"]) registry.register(orchestrator) registry.register(data_agent) # 门控 mcp_proxy = TrustProxy(default_min_trust=300, tool_policies={ "query_db": ToolPolicy(min_trust=500, required_capabilities=["query:db"]), }) a2a_gate = TrustGate(TrustPolicy(min_trust_score=400)) # Orchestrator → A2A → DataAgent → MCP → database tool task = TaskEnvelope.create(skill_id="query:db", source_did=str(orchestrator.did), source_trust_score=850, input_text="SELECT count(*) FROM transactions WHERE amount > 10000") if a2a_gate.evaluate(task).allowed: task.start() auth = mcp_proxy.authorize(agent_did=str(data_agent.did), agent_trust_score=800, agent_capabilities=["query:db"], tool_name="query_db") if auth.allowed: task.complete("1,247 high-value transactions")这条链路展示了本教程的核心思想:信任分在每一跳都被重新验证——A2A 层用TrustGate验证编排器、MCP 层用TrustProxy验证数据智能体,任何一跳不达标都会中断任务。
9. CLI——agentmesh proxy
启动一个透明的 MCP 代理,为任意 MCP 服务器包裹治理能力。
9.1 基本用法
# 用严格策略代理 filesystem MCP 服务器 agentmesh proxy --policy strict \ --target npx --target -y \ --target @modelcontextprotocol/server-filesystem --target /Users/me # 用中等策略、不带验证页脚代理 Python MCP 服务器 agentmesh proxy --policy moderate --no-footer \ --target python --target my_mcp_server.py9.2 CLI 选项
| 选项 | 说明 | 默认值 |
|---|---|---|
--target,-t | 目标 MCP 服务器命令(可重复以追加参数) | 必填 |
--policy,-p | 策略级别:strict、moderate、permissive | strict |
--no-footer | 在输出中禁用验证页脚 | false |
--identity,-i | Agent 身份名称 | mcp-proxy |
9.3 策略级别
- Strict(默认):拦截写/删操作,拦截
/etc与/root路径,仅放行读操作; - Moderate:写操作给出告警但允许执行,拦截关键系统路径;
- Permissive:放行所有操作,全部记录用于审计。
9.4 目标二进制白名单
代理限制可被拉起的二进制。默认白名单:npx, node, python, python3, uvx, uv, echo, cat, test。通过环境变量扩展:
export AGENTMESH_PROXY_ALLOWED_TARGETS=ruby,cargo,deno9.5 代理工作原理
MCPProxy类(位于agentmesh.cli.proxy)横亘在 stdin/stdout 之间,拦截 JSON-RPC 2.0 消息。每次tools/call的处理流程:
- 解析 JSON-RPC 请求(非 JSON 消息直接丢弃);
- 构建策略上下文(
tool、path); - 交给
PolicyEngine评估; - 被拦截→ 向客户端返回 JSON-RPC 错误(code
-32001); - 放行→ 转发给目标,信任分 +1;
- 响应→ 可选附加验证页脚;
- 写入
AuditLog。
10. 相关教程交叉引用
| 教程 | 关联关系 |
|---|---|
| Tutorial 02 — Trust and Identity | Agent DID、Ed25519 密钥、信任评分——所有协议桥的基础 |
| Tutorial 07 — MCP Security Gateway | MCPGateway与MCPSecurityScanner——互补的 MCP 保护 |
| Tutorial 01 — Policy Engine | MCPProxy与A2AGovernanceAdapter使用的 YAML 策略 |
| Tutorial 04 — Audit and Compliance | 跨协议交互的审计日志 |
| Tutorial 13 — Observability and Tracing | 经 IATPTracingContext传播的分布式追踪上下文 |
小结
| 概念 | 关键类 | 作用 |
|---|---|---|
| MCP 信任代理 | TrustProxy | 按工具的信任阈值、能力检查、速率限制与参数注入扫描 |
| MCP CLI 代理 | MCPProxy | 带策略执行的透明 stdin/stdout 代理 |
| A2A Agent Card | AgentCard | 带 AgentMesh 信任扩展的 A2A 兼容发现 |
| A2A 任务信封 | TaskEnvelope | 带状态机的信任验证任务生命周期 |
| A2A 信任门 | TrustGate | 入站 A2A 任务的策略执行(含按技能阈值覆盖) |
| A2A 治理 | A2AGovernanceAdapter | 带内容过滤的内核级 A2A 治理 |
| IATP 握手 | TrustHandshake | 带注册表验证的 Ed25519 挑战/响应 |
| IATP Proto | iatp.proto | 信任、认证、声誉的跨语言 gRPC 服务 |
| 可信 Agent Card | TrustedAgentCard | 密码学签名的智能体元数据卡片 |
| Card 注册表 | CardRegistry | 签名卡片发现与验证 |
| 信任桥 | TrustBridge | 带 HMAC 保护缓存的对端验证 |
| 协议桥 | ProtocolBridge | 跨协议消息翻译(A2A ↔ MCP ↔ IATP) |
下一步
- Tutorial 17 — Advanced Trust & Behavior:动态信任评分与行为分析;
- Tutorial 18 — Compliance Verification:自动化合规检查;
- Tutorial 44 — A2A Conversation Policy:用反馈环防护治理 A2A 会话。
【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit
创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考