Skip to content

Chapter 12 -- WoT Integration and Memory Security

Your Trails agent can do useful things. But can other agents find it? Can it protect the knowledge it accumulates? This chapter covers two systems that answer those questions: the Web of Things (WoT) integration layer that makes your agent discoverable by machines, and the memory security stack that prevents agents from spoofing, poisoning, or leaking shared knowledge.

Both systems are additive. An app that never publishes a WoT Thing Description works fine. An app that never uses the memory gateway still stores facts. But when you need discoverability and trust, both are ready -- and they compose naturally.

Learning objectives

After reading this chapter, you will be able to:

  • Model your agent as a WoT Thing with AgentCard
  • Serialize capabilities as W3C Thing Descriptions
  • Expose a /.well-known/wot discovery endpoint
  • Use the trails wot describe CLI to inspect your agent's TD
  • Stand up a MemoryGateway that enforces identity and confidence
  • Configure trust levels, data classifications, and read policies
  • Detect tampering with provenance hash chains
  • Control what facts cross federation boundaries

Why WoT?

The W3C Web of Things (WoT) standard defines a machine-readable format for describing what a device -- or in our case, an agent -- can do. A Thing Description (TD) lists actions, their inputs, their outputs, and security requirements. Any WoT-aware consumer (another agent, an IoT gateway, a discovery service) can parse the TD and know how to invoke your agent without reading your source code.

Trails maps each @capability to a td:ActionAffordance. The mapping is mechanical and happens at serialization time -- your capability code does not change. WoT is a projection of the same metadata that MCP uses, not a replacement (ADR-0015a).

AgentCard: modeling your agent as a WoT Thing

AgentCard is the central type. It collects all registered capabilities, converts them to WoT actions, and serializes the result as a W3C TD 1.1 JSON-LD document.

Building from the registry

The simplest path: let AgentCard scan the live capability registry.

from trails import capability
from trails.wot import AgentCard

@capability(id="notes.create", description="Create a note")
def create_note(ctx, title: str, body: str) -> dict:
    note = Note(title=title, body=body)
    ctx.kg.add(note)
    return {"id": note.id}

@capability(id="notes.search", description="Search notes", idempotent=True)
def search_notes(ctx, q: str) -> list:
    return Note.where(title__icontains=q).fetch(ctx)

# Build the card -- picks up both capabilities automatically
card = AgentCard.from_registry(
    base_iri="https://notes.example/agents/main",
    title="Notes Agent",
    description="A simple note-taking agent",
)
print(card)
# <AgentCard id='https://notes.example/agents/main' actions=2>

No tool list, no manual registration. Every @capability in the process is included.

With MCP endpoint

When your agent speaks MCP, pass the endpoint to get td:Form entries on each action. Consumers see exactly how to invoke capabilities over the MCP transport:

card = AgentCard.from_registry(
    base_iri="https://notes.example/agents/main",
    mcp_endpoint="mcp://localhost:8080",
)
td = card.to_td()
# Each action now has:
#   "forms": [{"href": "mcp://localhost:8080/tools/call",
#              "op": "invokeaction", ...}]

Direct construction (for tests)

When you need a card without a running registry -- unit tests, offline documentation -- construct it directly:

card = AgentCard(
    agent_id="did:web:example.org:agents:test",
    title="Test Agent",
    actions={
        "greet": {
            "description": "Say hello",
            "idempotent": True,
            "input": {"type": "object"},
        },
    },
)

Thing Descriptions: serialization

AgentCard offers two serialization methods.

to_td() -- W3C TD 1.1

Returns a dict conforming to the TD 1.1 structure. This is what /.well-known/wot serves and what WoT consumers expect:

import json

td = card.to_td()
print(json.dumps(td, indent=2))

{
  "@context": [
    "https://www.w3.org/2022/wot/td/v1.1",
    {
      "trails": "https://trails.dev/ont/core#",
      "prov": "http://www.w3.org/ns/prov#",
      "sh": "http://www.w3.org/ns/shacl#"
    }
  ],
  "@type": ["trails:AgentCard", "Thing"],
  "id": "https://notes.example/agents/main",
  "title": "Notes Agent",
  "securityDefinitions": {
    "biscuit_sc": {"scheme": "bearer", "format": "biscuit"}
  },
  "security": "biscuit_sc",
  "actions": {
    "notes.create": {
      "description": "Create a note",
      "idempotent": false
    },
    "notes.search": {
      "description": "Search notes",
      "idempotent": true
    }
  }
}

to_jsonld() -- full JSON-LD with extensions

A superset of to_td() that includes extra metadata you attach to card.extra. Use this when you need to advertise custom endpoints:

card.extra["trails:endpoint"] = "https://notes.example/mcp"
doc = card.to_jsonld()
# doc["trails:endpoint"] == "https://notes.example/mcp"

Capability-to-action mapping

Each Trails field maps to a WoT or namespaced field:

Trails field WoT field Notes
id action key Dict key, not a TD field
description td:description Direct
input_shape td:input (sh:node) SHACL IRI preserved
idempotent td:idempotent Boolean, TD 1.1 native
side_effects td:safe + trails:sideEffect Non-empty = safe: false
cost_estimate trails:costEstimate Trails extension
policy_required trails:policyRequired Trails extension
reasoning trails:reasoningMode none / rdfs / owl-rl

Fields that have no WoT equivalent ride as trails: namespaced extensions. The TD stays valid; WoT consumers that do not understand trails: simply ignore the extensions.

Discovery: the /.well-known/wot endpoint

The WoT Discovery module exposes two HTTP routes conforming to W3C WoT Discovery:

Method Path Content-Type Returns
GET /.well-known/wot application/td+json Host TD (directory)
GET /agents/{agent_id}/td application/td+json Individual agent TD

Mounting the router

from fastapi import FastAPI
from trails.wot_discovery import create_wot_router

app = FastAPI()
app.include_router(create_wot_router())

# Now:
#   GET /.well-known/wot         -> host TD with links to all capabilities
#   GET /agents/notes.create/td  -> individual TD for notes.create

How the host TD works

The host TD's links[] array enumerates all registered capabilities. Each link points to the individual agent TD endpoint:

from trails.wot_discovery import build_host_td

td = build_host_td()
# td["links"] == [
#   {"rel": "item", "href": "trails://myapp/agents/notes.create",
#    "type": "application/td+json"},
#   {"rel": "item", "href": "trails://myapp/agents/notes.search",
#    "type": "application/td+json"},
# ]

The base IRI is resolved from trails.toml ([project].base_iri), falling back to trails://<project.name>/, then trails://local.

Security definitions

Security is auto-detected:

  • If TRAILS_HTTP_TOKEN is set or [http].auth_token is configured, the TD uses bearer_sc (HTTP bearer).
  • Otherwise, nosec_sc (no security).

CLI: trails wot describe

Inspect your agent's TD from the command line without starting a server:

# Print the host TD
trails wot describe

# Print a specific agent's TD
trails wot describe --agent notes.create

# Override base IRI
trails wot describe --base-iri https://myapp.example

# Compact JSON (pipe-friendly)
trails wot describe --compact

# Filter with jq
trails wot describe | jq '.actions | keys'

# List all registered agents
trails wot list

Both commands auto-discover capabilities from the current project directory before generating output.

Example: publishing your agent to the WoT ecosystem

A complete working setup -- capabilities, AgentCard, discovery endpoint:

from fastapi import FastAPI
from trails import capability
from trails.wot import AgentCard
from trails.wot_discovery import create_wot_router

@capability(id="evidence.search",
            description="Search evidence by topic")
def search(ctx, topic: str) -> dict:
    hits = ctx.kg.match(labels=["Evidence"],
                        where={"topic": topic})
    return {"hits": hits}

@capability(id="evidence.summarize",
            description="Summarize evidence",
            idempotent=True)
def summarize(ctx, evidence_id: str) -> dict:
    return {"summary": "..."}

# Build and inspect the card
card = AgentCard.from_registry(
    base_iri="https://evidence.example/agents/main",
    title="Evidence Agent",
    description="Search and summarize clinical evidence",
    mcp_endpoint="https://evidence.example/mcp",
)

import json
print(json.dumps(card.to_td(), indent=2))

# Mount the discovery endpoint
app = FastAPI()
app.include_router(create_wot_router())

# Run: uvicorn app:app --port 8000
# Then: curl http://localhost:8000/.well-known/wot

A remote agent or IoT gateway can now fetch http://localhost:8000/.well-known/wot, parse the TD, discover evidence.search and evidence.summarize, read their input schemas, and invoke them via the MCP transport.

Why memory security matters

Agent memory is a high-value target. A compromised or misbehaving agent can:

  • Spoof identities -- claim another agent's DID to gain trust.
  • Inflate confidence -- set confidence: 1.0 to dominate recall.
  • Poison shared knowledge -- write false facts that other agents consume.
  • Leak confidential data -- export restricted facts to untrusted federation peers.
  • Hide taint -- inferences based on retracted facts remain trusted.

The memory security layer prevents these attacks at the framework level, transparently to agents using trails.invoke().

MemoryGateway: the trusted intermediary

All memory operations pass through MemoryGateway. The gateway sits between the agent and the MemoryStore and enforces six invariants:

from trails.memory_security import MemoryGateway, MemorySession

session = MemorySession(
    principal_did="did:key:alice",
    trust_level="authenticated",
)
gw = MemoryGateway(memory_store, session)

# Learn -- gateway injects identity, caps confidence, chains provenance
iri = gw.learn(
    "Deploy key rotates weekly",
    confidence_hint=0.9,
    topic="ops",
    source="observed in CI config",
    ttl_seconds=86400,  # expires in 24 hours
)

# Recall -- delegates to store, adds audit entry
facts = gw.recall("deploy key rotation", topic="ops")

# Correct -- enforces authorization for cross-agent corrections
gw.correct(iri, "Deploy key rotates daily", reason="updated schedule")

# Forget -- soft delete with full audit trail
gw.forget(iri, reason="no longer relevant")

What the gateway enforces on every operation:

  1. Identity injection. agent_did, timestamp, and provenance are set from the authenticated MemorySession. Agents cannot override these fields.
  2. Confidence calibration. confidence_hint is capped by the agent's trust level.
  3. Correction authorization. Cross-agent corrections require explicit Cedar policy.
  4. Provenance hash chain. Every operation appends a SHA-256-chained record for tamper evidence.
  5. Audit logging. Every operation is logged with agent DID, action, timestamp, and detail.
  6. TTL tracking. Facts with ttl_seconds are tracked for expiry-based garbage collection.

Identity enforcement: no spoofing

MemorySession carries the agent's DID (extracted from Biscuit token or MCP handshake) and trust level. The gateway reads these -- agents cannot set them:

from trails.memory_security import MemorySession

session = MemorySession(
    principal_did="did:key:z6Mk...",
    trust_level="established",
    roles=["analyst"],
)

session.trust_level_float  # 0.9
session.is_anonymous       # False

Anonymous sessions (did:key:anonymous or empty DID) receive the lowest trust multiplier (0.3). The gateway reads the session's principal_did for every learn, correct, and forget call -- there is no parameter that lets the agent override it.

Confidence calibration: trust but verify

An agent says "I am 95% confident." The gateway says "your trust level allows at most 70%." The lower value wins.

from trails.memory_security import calibrate_confidence, TrustLevel

# AUTHENTICATED agent capped at 0.7
calibrate_confidence(hint=0.95, agent_did="did:key:alice",
                     trust_level=TrustLevel.AUTHENTICATED)
# -> 0.7

# ESTABLISHED agent with high correction rate
calibrate_confidence(hint=0.8, agent_did="did:key:bob",
                     trust_level=TrustLevel.ESTABLISHED,
                     correction_rate=0.4)
# -> 0.54  (0.9 * max(0.5, 1.0 - 0.4) = 0.54)

Trust levels

Level Multiplier When assigned
ANONYMOUS 0.3 No DID or did:key:anonymous
AUTHENTICATED 0.7 DID-bearing agent (default)
ESTABLISHED 0.9 Agent with track record
HUMAN 1.0 Human operator
SYSTEM 1.0 Framework-internal operations

The formula: effective = min(hint, trust_level * max(0.5, 1.0 - correction_rate)). An agent whose facts are frequently corrected sees its effective trust decay toward 50% of its base level. This is automatic -- no manual intervention needed.

Provenance chains: tamper detection

Every memory operation appends to a SHA-256 hash chain. Each record's self_hash covers the record's identity fields plus the prev_hash of the preceding record:

from trails.memory_security import ProvenanceChain, verify_chain

chain = ProvenanceChain()

# Records are appended automatically by the gateway.
# Manual use for testing:
rec = chain.append(
    fact_iri="urn:trails:memory:fact:1",
    agent_did="did:key:alice",
    action="memory.learn",
    timestamp="2026-04-19T10:00:00+00:00",
)
print(rec.self_hash)   # sha256:a1b2c3...
print(rec.prev_hash)   # sha256:000...000 (genesis)

# Verify the entire chain
valid, broken = chain.verify()
assert valid
assert broken == []

# Or use the convenience wrapper
assert verify_chain(chain)

The gateway exposes verify_provenance() for on-demand verification:

valid, broken_indices = gw.verify_provenance()
if not valid:
    print(f"Chain broken at indices: {broken_indices}")
    # A CHAIN_BREAK audit event is logged automatically

If anyone tampers with a record (changes content, modifies a timestamp, reorders entries), the hash chain breaks and verify() reports the exact indices.

Trust levels: LOCAL, PEER, PUBLIC

trails.memory_trust.TrustLevel classifies where a fact originated. This is a different concept from the agent trust levels in memory_security -- those measure who wrote the fact, these measure where it came from:

from trails.memory_trust import TrustLevel

TrustLevel.LOCAL > TrustLevel.PEER    # True
TrustLevel.PEER > TrustLevel.PUBLIC   # True
Level Meaning
LOCAL Produced by agents on this instance
PEER Received from a federated trusted peer
PUBLIC From untrusted or external sources

Data classification: OPEN through RESTRICTED

DataClassification controls which trust levels may read a fact:

Classification Who can read Sensitivity
OPEN Any agent (LOCAL, PEER, PUBLIC) Lowest
INTERNAL LOCAL and PEER agents
CONFIDENTIAL LOCAL agents only
RESTRICTED Named agents only (explicit grant) Highest 
from trails.memory_trust import DataClassification, FactTrustMetadata

meta = FactTrustMetadata(
    trust_level=TrustLevel.LOCAL,
    classification=DataClassification.CONFIDENTIAL,
)

Topic-scoped read policies

TopicScopedReadPolicy restricts which agents can recall facts based on topic and classification:

from trails.memory_trust import TopicScopedReadPolicy, DataClassification

policy = TopicScopedReadPolicy(
    topic_rules={
        "financials": DataClassification.CONFIDENTIAL,
        "public-docs": DataClassification.OPEN,
    },
    default_classification=DataClassification.INTERNAL,
    agent_clearances={
        "did:key:auditor": {
            DataClassification.CONFIDENTIAL,
            DataClassification.INTERNAL,
        },
    },
)

# Check access
policy.can_read("did:key:auditor", "financials")   # True (explicit clearance)
policy.can_read("did:key:analyst", "financials")    # False (no clearance)
policy.can_read("did:key:analyst", "public-docs")   # True (OPEN)

# Filter a fact list -- drops facts the agent cannot see
visible = policy.filter_facts(all_facts, agent_did="did:key:analyst")

Cedar policies can be attached for fine-grained control. When provided, Cedar evaluation takes precedence over the default matrix:

policy = TopicScopedReadPolicy(
    cedar_policies=[{
        "effect": "permit",
        "principal": "did:key:analyst",
        "action": "memory.recall",
        "resource": {"type": "Trails::Memory::Fact", "topic": "financials"},
    }],
)

Contamination tracking

When a source fact is retracted or found unreliable, all downstream inferences built on it must be flagged. ContaminationTracker does this automatically by following supports links in the knowledge graph:

from trails.memory_trust import ContaminationTracker

tracker = ContaminationTracker(store=ctx.kg._store)

# Mark a fact as tainted
tracker.mark_tainted(
    "urn:trails:memory:fact:42",
    reason="Source paper retracted",
)

# Taint status
tracker.is_tainted("urn:trails:memory:fact:42")  # True

# All downstream facts linked via "supports" are transitively tainted
chain = tracker.taint_chain("urn:trails:memory:fact:42")
# ["urn:trails:memory:fact:43", "urn:trails:memory:fact:44"]

Taint propagation is recursive: if fact A supports fact B, and fact B supports fact C, tainting A taints both B and C.

Federation trust gates

FederationTrustGate controls which facts cross federation boundaries.

Export filtering

Facts are excluded from export if: - Their classification exceeds the peer's allowed level. - They are tainted. - Their confidence is below the configured floor. - They originated from the requesting peer (no echo).

from trails.memory_trust import FederationTrustGate, TrustLevel, DataClassification

gate = FederationTrustGate(
    local_did="did:key:my-instance",
    peer_trust_levels={
        "did:key:partner": TrustLevel.PEER,
        "did:key:public-api": TrustLevel.PUBLIC,
    },
    export_max_classification=DataClassification.INTERNAL,
    confidence_floor=0.3,
)

exportable = gate.filter_for_export(
    facts=all_facts,
    peer_trust_level=TrustLevel.PEER,
    fact_metadata=metadata_map,
)

Import validation

All imported facts receive trust metadata based on the source peer:

annotated = gate.validate_import(incoming_facts, source_peer="did:key:partner")
for fact, meta in annotated:
    print(f"Trust: {meta.trust_level.value}, tainted: {meta.tainted}")
Source trust Confidence multiplier Auto-tainted
LOCAL 1.0 No
PEER 0.7 No
PUBLIC 0.3 Yes

PUBLIC facts are automatically marked as tainted on import. The consuming agent can still use them, but downstream inferences inherit the taint.

Putting it together: WoT + secure memory

Here is a complete example combining WoT discoverability with secured multi-agent memory. Two agents share a memory store through gateways with different trust levels. The whole system is discoverable via WoT:

from fastapi import FastAPI
from trails import capability
from trails.wot import AgentCard
from trails.wot_discovery import create_wot_router
from trails.memory_security import (
    MemoryGateway,
    MemorySession,
    MemoryAuditLog,
    ProvenanceChain,
)
from trails.memory_trust import (
    TopicScopedReadPolicy,
    DataClassification,
    FederationTrustGate,
    TrustLevel,
)

# -- Shared infrastructure --

audit_log = MemoryAuditLog()
chain = ProvenanceChain()

# -- Agent A: senior analyst (established trust) --

session_a = MemorySession(
    principal_did="did:key:analyst-a",
    trust_level="established",
)
gw_a = MemoryGateway(
    memory_store, session_a,
    audit_log=audit_log,
    provenance_chain=chain,
)

# -- Agent B: junior agent (authenticated trust) --

session_b = MemorySession(
    principal_did="did:key:junior-b",
    trust_level="authenticated",
)
gw_b = MemoryGateway(
    memory_store, session_b,
    audit_log=audit_log,
    provenance_chain=chain,
)

# -- Capabilities --

@capability(id="research.learn",
            description="Store a research finding in memory")
def learn_finding(ctx, content: str, topic: str,
                  confidence: float = 0.8) -> dict:
    iri = gw_a.learn(content, confidence_hint=confidence,
                     topic=topic, source="research-agent")
    return {"iri": iri}

@capability(id="research.recall",
            description="Recall findings on a topic",
            idempotent=True)
def recall_findings(ctx, topic: str) -> list:
    return gw_a.recall(topic=topic)

# -- Read policy --

read_policy = TopicScopedReadPolicy(
    topic_rules={"clinical": DataClassification.CONFIDENTIAL},
    agent_clearances={
        "did:key:analyst-a": {
            DataClassification.CONFIDENTIAL,
            DataClassification.INTERNAL,
        },
    },
)

# -- Federation gate --

gate = FederationTrustGate(
    local_did="did:key:my-instance",
    export_max_classification=DataClassification.INTERNAL,
    confidence_floor=0.5,
)

# -- WoT discovery --

card = AgentCard.from_registry(
    base_iri="https://research.example/agents/main",
    title="Research Agent",
    description="Store and recall research findings with trust controls",
    mcp_endpoint="https://research.example/mcp",
)

app = FastAPI()
app.include_router(create_wot_router())

# -- Use it --

# Agent A stores a high-confidence fact
iri = gw_a.learn(
    "Patient cohort shows 15% improvement",
    confidence_hint=0.95,
    topic="clinical",
    source="RCT results 2026-Q1",
    ttl_seconds=7776000,  # 90 days
)
# Effective confidence: min(0.95, 0.9) = 0.9 (ESTABLISHED cap)

# Agent B tries to store with inflated confidence
iri_b = gw_b.learn(
    "Secondary analysis confirms trend",
    confidence_hint=0.99,
    topic="clinical",
)
# Effective confidence: min(0.99, 0.7) = 0.7 (AUTHENTICATED cap)

# Verify provenance integrity
valid, broken = chain.verify()
assert valid

# Check audit trail
recent = audit_log.query(action="memory.learn", limit=10)
print(f"{len(recent)} recent learn operations")

The agent is now: - Discoverable via GET /.well-known/wot - Secure via gateway-enforced identity and confidence - Auditable via provenance chain and audit log - Federation-ready via trust gates and data classification

What's next

This chapter introduced the WoT and memory security systems. For the full API reference and advanced configuration: