Zero-Trust Identity and Trust Scoring for AI Agents

from agentmesh import AgentIdentity

# Create an agent with a human sponsor
agent = AgentIdentity.create(
    name="DataProcessor",
    sponsor="alice@company.com",
    capabilities=["read:data", "write:reports"],
    organization="Analytics",
)

print(agent.did)          # did:mesh:a1b2c3d4e5f6...
print(agent.public_key)   # Base64-encoded Ed25519 public key
print(agent.status)       # "active"

# Sign data with the private key
signature = agent.sign(b"payload to authenticate")

# Verify the signature using only the public key
is_valid = agent.verify_signature(b"payload to authenticate", signature)
print(is_valid)  # True

# Export as JWK (JSON Web Key) for interoperability
jwk = agent.to_jwk(include_private=False)

# Export as a W3C DID Document
did_doc = agent.to_did_document()

from agentmesh import AgentDID

did = AgentDID.generate()
print(did)  # did:mesh:7f3a9b2c1d4e5f6a...

# Parse an existing DID string
did = AgentDID.from_string("did:mesh:7f3a9b2c...")
print(did.method)     # "mesh"
print(did.unique_id)  # "7f3a9b2c..."

{
  "@context": ["https://www.w3.org/ns/did/v1"],
  "id": "did:mesh:<unique-id>",
  "verificationMethod": [{
    "id": "did:mesh:<unique-id>#<key-id>",
    "type": "Ed25519VerificationKey2020",
    "controller": "did:mesh:<unique-id>",
    "publicKeyBase64": "<base64-public-key>"
  }],
  "authentication": ["did:mesh:<unique-id>#<key-id>"],
  "service": [{
    "id": "did:mesh:<unique-id>#agentmesh",
    "type": "AgentMeshIdentity",
    "serviceEndpoint": "https://mesh.agentmesh.dev/v1"
  }]
}

spiffe://{trust_domain}/agentmesh[/{organization}]/{agent_name}

from agentmesh import HumanSponsor

sponsor = HumanSponsor.create(
    email="alice@company.com",
    name="Alice",
    organization="Analytics",
    allowed_capabilities=["read:data", "write:reports", "execute:analysis"],
)

sponsor.verify(method="email")

print(sponsor.max_agents)           # 10
print(sponsor.max_delegation_depth) # 3

if sponsor.can_sponsor_agent():
    agent = AgentIdentity.create(
        name="NewAgent",
        sponsor=sponsor.email,
        capabilities=["read:data"],
    )

from agentmesh import RiskScorer
from agentmesh.identity.risk import RiskSignal

scorer = RiskScorer()

# Get or create a score for an agent
score = scorer.get_score("did:mesh:abc123...")
print(score.total_score)  # 500 (default)
print(score.risk_level)   # "medium"

# Report a risk signal (value: 0.0 = no risk, 1.0 = maximum risk)
scorer.add_signal(
    agent_did="did:mesh:abc123...",
    signal=RiskSignal(
        signal_type="behavior.anomaly",
        severity="high",
        value=0.8,          # 0.0 = no risk, 1.0 = max risk
        source="anomaly_detector",
        details="Unusual data access pattern detected",
    ),
)

# Score recalculates automatically
updated_score = scorer.recalculate("did:mesh:abc123...")

# Find agents that have decayed into high-risk territory
high_risk_agents = scorer.get_high_risk_agents(threshold=400)

# Parent agent with broad capabilities
parent = AgentIdentity.create(
    name="OrchestratorAgent",
    sponsor="alice@company.com",
    capabilities=["read:data", "write:reports", "execute:analysis"],
)

# Delegate a narrower set to a child
child = parent.delegate(
    name="ReportWriter",
    capabilities=["write:reports"],  # Must be a strict subset of parent's
)

print(child.parent_did)                        # Parent's DID
print(child.delegation_depth)                  # 1
print(child.has_capability("write:reports"))   # True
print(child.has_capability("read:data"))       # False — not delegated

from agentmesh import ScopeChain

# Create a root chain from a human sponsor
chain, root_link = ScopeChain.create_root(
    sponsor_email="alice@company.com",
    root_agent_did="did:mesh:root...",
    capabilities=["read:data", "write:reports", "execute:analysis"],
    sponsor_verified=True,
)

# Verify the entire chain is intact
is_valid, error = chain.verify()
print(is_valid)  # True

# Trace how a specific capability was granted
trace = chain.trace_capability("write:reports")
# Returns the full delegation path for that capability

from agentmesh import TrustBridge

bridge = TrustBridge(
    agent_did=str(agent_a.did),
    default_trust_threshold=700,
)

bridge.register_peer(
    peer_did="did:mesh:agent_b...",
    peer_name="DataAnalyzer",
    protocol="iatp",
)

# Verify a peer (uses cached result if available)
result = asyncio.run(bridge.verify_peer(
    peer_did="did:mesh:agent_b...",
    required_trust_score=700,
    required_capabilities=["read:data"],
))

# Quick trust check
is_trusted = asyncio.run(bridge.is_peer_trusted(
    peer_did="did:mesh:agent_b...",
    required_score=700,
))

from agentmesh import CredentialManager

manager = CredentialManager(default_ttl=900)  # 15 minutes

# Issue a scoped credential
cred = manager.issue(
    agent_did="did:mesh:abc123...",
    capabilities=["read:data"],
    resources=["dataset_sales", "dataset_inventory"],
    ttl_seconds=900,
)

print(cred.credential_id)    # "cred_a1b2c3..."
print(cred.status)            # "active"
print(cred.to_bearer_token()) # "Bearer <token>"

# Validate an incoming token (constant-time comparison)
incoming_token = request.headers["Authorization"].removeprefix("Bearer ")
cred = manager.validate(incoming_token)

if cred and cred.is_valid():
    if cred.has_capability("read:data"):
        if cred.can_access_resource("dataset_sales"):
            # Authorized — proceed
            pass

# Rotate before expiry (threshold: 60 seconds)
cred = manager.rotate_if_needed(cred.credential_id)

# Revoke immediately on compromise
manager.revoke(cred.credential_id, reason="Suspected compromise")

# Revoke ALL credentials for a compromised agent
count = manager.revoke_all_for_agent(
    agent_did="did:mesh:compromised...",
    reason="Agent suspended pending investigation",
)