Capabilities & Dispatch¶

Capabilities are the primary unit of functionality in Trails. Each capability is a named, typed, provenance-tracked function that can be discovered and invoked via MCP, HTTP, or direct Python calls.

The @capability decorator¶

Register a function as a Trails capability with @capability. Three call forms are supported:

Bare form — id inferred from function name¶

from trails import capability

@capability
def greet(name: str) -> dict:
    return {"message": f"Hello, {name}!"}

The capability id becomes "greet" (the function's __name__).

Positional id¶

@capability("user.greet")
def greet(name: str) -> dict:
    return {"message": f"Hello, {name}!"}

Keyword id (with optional alias name=)¶

@capability(id="user.greet", description="Greet a user by name")
def greet(name: str) -> dict:
    return {"message": f"Hello, {name}!"}

name= is an alias for id=. Passing both with different values raises TrailsError.

Decorator parameters¶

Deprecated (no-op): reasoning= and assurance= are accepted for one deprecation cycle but emit DeprecationWarning and are otherwise ignored. Reasoning mode is feature-detected from the loaded ontology (ADR-0021); assurance lives on the provenance writer.

Validation rules at decoration time¶

• The id must be a non-empty string with no whitespace. Violations raise TrailsError with a suggested fix.
• async def handlers are rejected in M0 (sync-only; async lands in M1).
• Registering a duplicate id raises TrailsError pointing to the first registration site.
• A leading ctx parameter is stripped from the recorded param list (it is injected by the runtime, not supplied by the caller).
• The original function is returned unchanged — user code can still call it directly outside invoke().

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

When input_shape is omitted, the decorator inspects the handler's type annotations. If the first non-ctx parameter is annotated with a @shape-decorated class, its shape IRI is used automatically.

Before (explicit input_shape)¶

from trails import capability, shape, predicate

@shape("https://example.org/QueryInput")
class QueryInput:
    query = predicate("schema:query", str, min_count=1)

@capability(
    "answer_query",
    input_shape="https://example.org/QueryInput",   # manual
    description="Answer a question",
)
def answer_query(q: QueryInput) -> dict:
    return {"answer": "..."}

After (shape-from-annotation)¶

@shape("https://example.org/QueryInput")
class QueryInput:
    query = predicate("schema:query", str, min_count=1)

@capability("answer_query", description="Answer a question")
def answer_query(q: QueryInput) -> dict:      # ← input_shape auto-derived
    return {"answer": "..."}

The decorator calls _derive_shape_iri(fn), which walks the handler's parameters (skipping a leading ctx), checks each annotation for _trails_shape metadata, and returns the first matching IRI. If no annotation carries shape metadata, input_shape stays None.

invoke() — capability dispatch¶

Call a registered capability through the runtime:

from trails import invoke

result = invoke("greet", {"name": "Ada"})

Signature¶

def invoke(
    id: str,
    args: dict[str, Any] | None = None,
    *,
    principal: str = "did:local:m0",
    principal_attrs: dict[str, Any] | None = None,
) -> dict:

Parameters¶

Dispatch lifecycle¶

invoke("answer_query", {"query": "..."})
   │
   ├─ 1. Resolve capability from registry
   ├─ 2. Check required args are present
   ├─ 3. Soft-validate input against input_shape (M1 warnings; M2 hard)
   ├─ 4. Mint trace_id (UUIDv4 stopgap; UUIDv7 in M1)
   ├─ 5. Build Context (if handler declares `ctx`)
   ├─ 6. Evaluate @policy (if present) → permit / deny
   ├─ 7. Run @before middleware → may update args
   ├─ 8. Run @around middleware stack → wraps handler
   ├─ 9. Execute handler function
   ├─ 10. Run @after middleware → may replace result
   ├─ 11. Soft-validate output against output_shape
   ├─ 12. Attach PROV-O provenance via kernel Store
   └─ 13. Return response envelope

Steps 6 is enforced when @policy is declared. Steps 3 and 11 use soft validation (log warnings) in M1, hard validation from M2+.

Return value¶

invoke() returns a response envelope dict:

{
    "payload": {"message": "Hello, Ada!"},
    "provenance": { ... },     # PROV-O activity metadata
    "capability": "greet",
    "trace_id": "a1b2c3d4-..."
}

The provenance block contains the PROV-O activity record linking the capability, its inputs, and its outputs. The trace_id is a UUIDv4 assigned to this invocation for end-to-end tracing.

Error handling¶

• Unknown capability id: TrailsError with "did you mean?" suggestions from known capabilities.
• Missing required args: TrailsError listing missing, provided, and expected parameters.
• Handler exceptions: wrapped in TrailsError chained via from exc. Framework errors (TrailsError) propagate directly.
• Non-JSON-serializable results: TrailsError.

Context injection¶

When a handler's first parameter is named ctx, the runtime builds a fresh Context object per invocation and passes it positionally:

@capability("lookup")
def lookup(ctx, entity_iri: str) -> dict:
    results = ctx.kg.query(
        f"SELECT ?p ?o WHERE {{ <{entity_iri}> ?p ?o }}"
    )
    return {"triples": results}

Handlers without a ctx parameter receive only **args — no breakage for simple capabilities.

What ctx provides¶

ctx.kg — the graph store handle¶

The KG object provides both ORM-style and raw SPARQL access:

ctx.kg.query() accepts an optional reason=True to trigger automatic OWL-RL / RDFS materialization before executing the query (ADR-0021 progressive enhancement).

Provenance recording¶

Every capability invocation generates a PROV-O activity record:

_:activity a prov:Activity ;
    prov:wasAssociatedWith <urn:trails:capability:answer_query> ;
    prov:startedAtTime "2026-04-12T10:30:00Z"^^xsd:dateTime ;
    prov:endedAtTime "2026-04-12T10:30:01Z"^^xsd:dateTime ;
    prov:used _:input ;
    prov:generated _:output .

Provenance is stored in a named graph (urn:trails:prov) in the same store. The handler result must be JSON-serializable — invoke() calls json.dumps(result) and passes the JSON to the kernel's attach_provenance() FFI method.

Async capabilities¶

Async handlers (async def) are not supported in M0. Decorating an async def function raises TrailsError at decoration time:

# This raises TrailsError at import time:
@capability("fetch")
async def fetch(url: str) -> dict:   # ← rejected in M0
    ...

Async support (including an async invoke() path) is planned for M1 as part of the DispatchCoordinator evolution.

Middleware hooks¶

Cross-cutting concerns (logging, metrics, input/output transforms) can be attached via middleware decorators without modifying the capability body. Four hooks are available:

All hooks accept a capability id or glob pattern ("notes.*", "*" for catch-all). Multiple hooks of the same kind run in registration order; @around forms a LIFO stack (last registered = outermost wrapper).

See Middleware for the full API, execution order, and error-handling semantics.

Pipeline pattern for request processing¶

The Pipeline class (in trails.pipeline) implements Chain of Responsibility for /invoke request processing. Each concern (auth, rate limiting, validation, policy, execution, provenance, cost, observability) is an independent stage that receives a shared PipelineContext and either processes it or aborts the pipeline.

from trails.pipeline import default_pipeline, Stage, PipelineContext

# Start with the built-in stage chain.
pipe = default_pipeline()

# Add a custom stage after auth.
class AuditStage(Stage):
    name = "audit"
    def process(self, ctx: PipelineContext) -> None:
        print(f"Invoking {ctx.capability_id} as {ctx.principal}")

pipe.insert_after("auth", AuditStage())
result = pipe.execute(ctx)

Built-in stages: AuthStage, RateLimitStage, ValidationStage, PolicyStage, InvokeStage, ProvStage, CostStage, ObservabilityStage. See pipeline.py for the full API.

Examples¶

Example 1: simple hello-world¶

from trails import capability, invoke

@capability("greet", description="Greet a user")
def greet(name: str) -> dict:
    return {"message": f"Hello, {name}!"}

result = invoke("greet", {"name": "Ada"})
print(result["payload"])
# {"message": "Hello, Ada!"}
print(result["trace_id"])
# "a1b2c3d4-e5f6-..."

Example 2: shape annotation with context¶

from trails import capability, shape, predicate, invoke

@shape("https://example.org/QueryInput")
class QueryInput:
    query = predicate("schema:query", str, min_count=1)

@shape("https://example.org/QueryResult")
class QueryResult:
    answer = predicate("schema:answer", str, min_count=1)
    sources = predicate("schema:citation", str)

@capability(
    "answer_query",
    output_shape="https://example.org/QueryResult",
    description="Answer a question using the knowledge graph",
)
def answer_query(ctx, q: QueryInput) -> dict:
    # input_shape is auto-derived from the QueryInput annotation.
    # ctx.kg is available for graph queries.
    rows = ctx.kg.query(
        f'SELECT ?o WHERE {{ ?s schema:answer ?o }}'
    )
    return {
        "answer": rows[0]["o"] if rows else "Unknown",
        "sources": ["https://example.org/doc/1"],
    }

result = invoke("answer_query", {"q": {"query": "What is Trails?"}})

In this example:

• input_shape is omitted from @capability(...) — the decorator auto-derives it from the QueryInput type annotation on parameter q.
• output_shape is still declared explicitly.
• ctx is injected by the runtime and provides ctx.kg for graph access.

Example 3: bare decorator with middleware¶

from trails import capability, invoke
from trails.decorators import before, after

@before("audit")
def log_call(ctx, args):
    print(f"Calling audit with {args}")

@after("audit")
def tag_result(ctx, args, result):
    result["audited"] = True
    return result

@capability
def audit(record_id: str) -> dict:
    return {"status": "reviewed", "record": record_id}

result = invoke("audit", {"record_id": "R-42"})
# prints: Calling audit with {'record_id': 'R-42'}
# result["payload"]["audited"] == True