Tutorial 16 — Protocol Bridges (A2A, MCP, IATP)¶

from mcp_trust_proxy import TrustProxy, ToolPolicy

# Create a trust proxy with per-tool policies
proxy = TrustProxy(
    default_min_trust=300,
    tool_policies={
        "file_write": ToolPolicy(min_trust=800, required_capabilities=["fs_write"]),
        "shell_exec": ToolPolicy(blocked=True),
    },
)

# Agent requests tool access
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 default threshold
print(result.reason)        # "Authorized"
print(result.trust_score)   # 600

┌──────────────┐     ┌─────────────────┐     ┌──────────────┐
│  MCP Client  │ ──► │  AgentMesh      │ ──► │  MCP Server  │
│  (Claude,    │ ◄── │  Trust Proxy    │ ◄── │  (filesystem,│
│   VS Code)   │     │                 │     │   database)  │
└──────────────┘     └─────────────────┘     └──────────────┘
                       ● Policy check      ● Rate limiting
                       ● Trust score gate  ● Audit logging

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))

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}}

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, ...}

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}

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())

{
  "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"]

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

wire_format = task.to_dict()
# Includes: id, status.state, skill_id, messages[], x-agentmesh-trust{}
received = TaskEnvelope.from_dict(wire_format)

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,
))

# Evaluate an incoming task
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"

# Auto-fail denied tasks
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

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

# Blocked content
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..."

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    1000

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"],
)

# Create a signed card from identity
card = TrustedAgentCard.from_identity(identity)
print(card.agent_did)        # "did:mesh:..."
print(card.card_signature)   # Base64 Ed25519 signature

# Verify — anyone with public key can check
assert card.verify_signature()

# Tamper → verification fails
card.name = "MaliciousAgent"
assert not card.verify_signature()

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 check

from agentmesh.trust.bridge import TrustBridge

bridge = TrustBridge(
    agent_did="did:mesh:my-agent", default_trust_threshold=700,
    identity=my_identity, registry=identity_registry,
)

# Verify peer (IATP handshake), result is HMAC-cached
result = await bridge.verify_peer(
    peer_did="did:mesh:remote-agent", protocol="iatp",
    required_trust_score=600, required_capabilities=["read:data"],
)

# Quick cached check (verifies HMAC integrity of stored record)
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")

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"]

# 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)

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)

from agentmesh.identity import AgentIdentity, IdentityRegistry
from a2a_agentmesh import TaskEnvelope, TrustGate, TrustPolicy
from mcp_trust_proxy import TrustProxy, ToolPolicy

# Identities
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)

# Gates
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")

# Proxy a filesystem MCP server with strict policy
agentmesh proxy --policy strict \
  --target npx --target -y \
  --target @modelcontextprotocol/server-filesystem --target /Users/me

# Proxy a Python MCP server with moderate policy, no footers
agentmesh proxy --policy moderate --no-footer \
  --target python --target my_mcp_server.py