Growing Your KG App¶

You ran trails new blog && cd blog && trails server. It said "ready". Now what? This tutorial grows that scaffold into an app with typed data, validation, reasoning, and policy — in that order, adding one feature at a time. You only touch RDF in Step 4, after three steps of real work.

The design rationale lives in ADR-0021: one surface, features are additive, you never pick a tier. Code you write in Step 1 still works unchanged after Step 7.

Step 0: What you have¶

trails new blog gave you two files:

blog/
  app.py
  trails.toml   # optional; delete it and Trails infers from the dirname

app.py holds one @capability. That's it. There is no shapes/, no ontology/, no policies/ directory. Those aren't missing — they're not needed yet. You'll add each one in the step that first requires it, and not before. Every file you don't have is a file you don't have to read, debug, or keep in your head.

Step 1: Bare capability (hello world)¶

Open app.py. The scaffold has a hello capability. Add a second function next to it:

from trails import capability

@capability
def hello(ctx):
    return {"msg": "hi"}

@capability
def greet(ctx, name: str):
    return {"msg": f"hello, {name}"}

Reload (trails server picks up changes). Both capabilities are now dispatchable; one can invoke the other via ctx.invoke("greet", name=...).

The capability id is inferred from the function name (hello, greet). Override it by passing a positional string:

@capability("blog.greet")
def greet(ctx, name: str):
    return {"msg": f"hello, {name}"}

Or with keyword arguments when you want metadata:

@capability(id="blog.greet", description="Greet a user by name")
def greet(ctx, name: str):
    return {"msg": f"hello, {name}"}

All three forms — bare @capability, positional id @capability("id"), and keyword @capability(id=..., description=...) — produce the same result. Use whichever reads best. No new concepts, no config change, no ontology. Two functions, two capabilities.

Step 2: Add a node type with typed data¶

You want a create_note capability that takes a note object. Introduce @node_type — this declares a data shape with named fields:

from trails import capability, node_type

@node_type("Note", fields={"title": str, "content": str})
class Note: ...

@capability
def create_note(ctx, title: str, content: str):
    note = Note(title=title, content=content)
    ctx.kg.add(note)
    return {"id": note.id}

@node_type("Note", fields={...}) tells Trails: writes labelled Note must carry a title (str) and content (str). Mismatches are rejected at write time with a readable error. The kernel stores the node with a Note label and the declared fields as properties — no IRI, no predicate, no turtle file.

ctx.kg.add(note) persists the instance through the kernel store. The minted IRI is available as note.id immediately after construction.

Notice what did not change: hello and greet are untouched. They don't know Note exists. @node_type is purely additive — a new label gets validation; every other capability keeps working exactly as before. No migration, no config flag, no tier switch.

Supported field types¶

@node_type accepts these types in the fields={} dict:

Linking two node types¶

Declare a second @node_type and point at it from the first — the field becomes a typed edge (an IRI-to-IRI triple, not a literal):

@node_type("Author", fields={"name": str})
class Author: ...

@node_type("Note", fields={"title": str, "author": Author})
class Note: ...

@capability
def notes_by(ctx, author_name: str):
    return [n.title for n in Note.where(author__name=author_name).fetch(ctx)]

Declaration order matters — the target class must exist before the referencing class. author hydrates back as an IRI string (no auto-traversal); follow it explicitly with Author.find(ctx, note.author) when you need the target object.

The chain in where() does the hop server-side in one SPARQL BGP, so author__name="Alice" is the right move over a per-note find() loop. Multi-valued edges use list[Author] with identical semantics.

Step 3: Add @shape for validation constraints¶

You want stricter rules: title is required, at most 120 characters; content is at least one character. @node_type gives you types; @shape gives you constraints.

from trails import shape, predicate, node_type

@node_type("Note", fields={"title": str, "content": str})
@shape
class Note:
    title   = predicate("schema:name", required=True, max_length=120)
    content = predicate("schema:description", required=True, min_length=1)

Both @node_type and @shape coexist on the same class. @shape layers SHACL-grade constraint checks on top of the type declarations. If a write violates a constraint, the caller sees:

SHACL violation on Note.title: sh:maxLength failed — expected length <= 120, got 186 (value="...")

One line describing the problem. No stack trace in the user's face.

@shape call forms¶

Like @capability, @shape supports three forms:

# Bare — IRI is auto-minted from trails.toml project name + class name
@shape
class Note: ...

# Positional IRI
@shape("https://myapp.example/ns/Note")
class Note: ...

# Keyword IRI
@shape(iri="https://myapp.example/ns/Note", prefixes={"schema": "http://schema.org/"})
class Note: ...

predicate() options¶

The predicate() function declares an RDF predicate binding with cardinality and value constraints:

predicate(
    "schema:name",          # predicate IRI
    required=True,          # alias for min=1 (sh:minCount)
    many=False,             # alias for max=1 (sh:maxCount); many=True → unbounded
    min_length=1,           # sh:minLength (strings)
    max_length=120,         # sh:maxLength (strings)
    min_value=0,            # sh:minInclusive (numerics)
    max_value=100,          # sh:maxInclusive (numerics)
    pattern=r"^[A-Z]",     # sh:pattern (regex, validated at declaration time)
    one_of=["a", "b", "c"],# sh:in (enumeration of allowed values)
)

All constraint kwargs are optional and additive — a predicate() call that doesn't pass them behaves exactly like a plain type declaration.

Shape-from-annotation (auto-derived input_shape)¶

When a @capability handler's first non-ctx parameter is annotated with a @shape-decorated class, the framework auto-derives the input_shape — no need to pass it explicitly:

@shape("https://example.org/ns/NoteInput")
class NoteInput:
    title = predicate("schema:name", required=True)

@capability
def create_note(ctx, note: NoteInput):
    # input_shape is auto-derived from NoteInput's @shape IRI
    ...

Step 4: Ontology and reasoning¶

You want to query across note-like things — Note, future Draft, future Comment — as one family, and you want a standard vocabulary so a third-party MCP client understands what you publish. This is where RDF enters, for the first time in this tutorial. Three steps of real work have already happened. That's the point.

Create ontology/notes.ttl:

@prefix schema: <http://schema.org/> .
@prefix app:    <https://example.org/blog/> .

app:Note rdfs:subClassOf schema:CreativeWork .

Note is now declared a subclass of schema:CreativeWork. With reasoning on, a query for schema:CreativeWork also returns notes:

SELECT ?x WHERE { ?x a schema:CreativeWork }

Without reasoning, this query returns nothing: you never labelled a node schema:CreativeWork. With reasoning, the rdfs:subClassOf edge from app:Note pulls every Note into the result set. Add Draft as another subclass later and the query continues to work — the ontology is where taxonomy lives, not the queries.

This is the first RDF in the tutorial — three capabilities and a whole validation layer came first. The same Python code from Step 3 works unchanged; the ontology is a separate file, read by the reasoner, never referenced from app.py. You pay for reasoning only on queries that request it; writes stay as fast as Step 2.

Step 5: Add @policy for access control¶

You want to restrict create_note to authenticated users. The @policy decorator attaches Cedar policy metadata:

from trails import capability, policy

@capability
@policy("notes.cedar::permit_authenticated")
def create_note(ctx, title: str, content: str):
    note = Note(title=title, content=content)
    ctx.kg.add(note)
    return {"id": note.id}

Decorator ordering matters: @capability must be outermost, @policy directly below it. Reversed ordering raises TrailsError with a fix hint.

The Cedar policy file notes.cedar lives next to app.py (path resolved relative to the handler's source file):

// notes.cedar
permit(
    principal,
    action == Action::"create_note",
    resource
)
when { principal.role == "author" };

forbid(
    principal,
    action == Action::"create_note",
    resource
)
unless { principal.role == "author" };

Cedar head clauses¶

Trails supports a deliberate subset of Cedar:

• Principal: principal, principal == User::"alice", principal in Group::"admins"
• Action: action, action == Action::"create_note", action in [Action::"create_note", Action::"update_note"]
• Resource: resource, resource == Note::"123", resource is Trails::Resource::Note

Body blocks use when { ... } and unless { ... } with == / != conditions joined by &&.

Principal attributes¶

For role-based policies, register principal attributes at app startup:

from trails import register_principal_attrs

register_principal_attrs("alice", {"role": "author"})
register_principal_attrs("bob", {"role": "reader"})

The policy engine resolves principal.role against these attribute dicts at evaluation time.

Step 6: Add middleware (@before / @after)¶

Cross-cutting concerns — logging, metrics, input transformation — register without touching the capability body. Four middleware decorators are available:

from trails import before, after, on_error, around

@before("create_note")
def attach_timestamp(ctx, args):
    """Inject a created_at timestamp into the args dict."""
    import time
    return {"created_at": time.time()}

@after("create_note")
def log_creation(ctx, args, result):
    """Log every note creation. Return None to keep result unchanged."""
    print(f"Created note: {result}")
    return None

@on_error("create_note")
def handle_error(ctx, args, exc):
    """Run when the handler raises. Return None to re-raise."""
    print(f"Error creating note: {exc}")
    return None

@around("*")
def timing_wrapper(ctx, args, next):
    """Wrap every capability with timing. Call next() exactly once."""
    import time
    t0 = time.monotonic()
    result = next()
    elapsed = time.monotonic() - t0
    print(f"Capability took {elapsed:.3f}s")
    return result

Patterns¶

All middleware decorators accept a capability id or a glob pattern:

Execution order¶

Per invocation:

1. @around wraps the entire invocation (LIFO stack — last registered is outermost).
2. Inside the innermost @around, all matching @before run in registration order.
3. The handler runs. 4a. On success: all matching @after run in registration order. 4b. On failure: all matching @on_error run in registration order.

@before may return a dict to merge into args. @after may return a value to replace result. @on_error may return an Exception to replace the current one.

Step 7: Add ORM queries (.where(), .values(), Q)¶

The @node_type ORM surface supports Django-style querying via Model.where() and QueryBuilder:

Basic queries¶

@capability
def list_notes(ctx):
    """Return all notes."""
    return [{"id": n.id, "title": n.title} for n in Note.where().fetch(ctx)]

@capability
def recent_notes(ctx, min_priority: int):
    """Filter with Django-style suffixes."""
    notes = Note.where(priority__gte=min_priority).fetch(ctx)
    return [{"id": n.id, "title": n.title} for n in notes]

Filter suffixes¶

.values() projection¶

Project to plain dicts instead of hydrated model instances:

@capability
def note_titles(ctx):
    """Return only id + title, no full hydration."""
    return Note.where().values("title").fetch(ctx)
    # → [{"id": "trails://...", "title": "Hello"}, ...]

Q objects for complex logic¶

Q composes filters with | (OR), & (AND), and ~ (NOT):

from trails.orm import Q

@capability
def search_notes(ctx, query: str):
    """OR search across title and content."""
    notes = Note.where(
        Q(title__icontains=query) | Q(content__icontains=query)
    ).fetch(ctx)
    return [{"id": n.id, "title": n.title} for n in notes]

@capability
def active_notes(ctx):
    """NOT + AND composition."""
    notes = Note.where(
        ~Q(status="archived") & Q(priority__gte=3)
    ).fetch(ctx)
    return [{"id": n.id, "title": n.title} for n in notes]

Property-path traversal¶

Django-style double-underscore chains traverse reference fields server-side in one SPARQL BGP:

# 1-hop: notes by a specific author name
Note.where(author__name="Alice").fetch(ctx)

# 2-hop: notes by an author in a specific org
Note.where(author__org__name="Acme").fetch(ctx)

ctx.kg escape hatches¶

For queries beyond what the ORM covers, use ctx.kg directly:

@capability
def label_first_workflow(ctx):
    """Create and query label-first nodes (no @node_type needed)."""
    # Create a node with labels and properties
    iri = ctx.kg.node(labels=["Bookmark"], properties={"url": "https://example.com"})

    # Create an edge between two nodes
    ctx.kg.edge(subject=iri, label="tagged_with", object=tag_iri)

    # Match nodes by labels and/or types
    results = ctx.kg.match(labels=["Bookmark"], where={"url": "https://example.com"})

    # Raw SPARQL query
    rows = ctx.kg.query("SELECT ?s ?title WHERE { ?s <pred> ?title }")

    # Raw SPARQL update
    ctx.kg.update("INSERT DATA { <s> <p> <o> }")

What you built¶

You added seven features, in the order you actually needed them:

Nothing you wrote in Step 1 was wasted; nothing had to be rewritten to make room for what came later. That is the progressive-enhancement promise.

Where next¶

• Capabilities & Dispatch — the full @capability surface, ctx.invoke, provenance.
• Shapes & Validation — @node_type, @shape, predicate(), SHACL export.
• Policy & Authorization — Cedar policies, cost envelopes, @policy decorator.
• MCP Integration — exposing your capabilities to Claude Desktop and other MCP clients.
• ADR-0021 — why the surface is one surface, and why features are strictly additive.