- Use
domain.verb_nounformat:billing.list_invoices,users.get_profile. - Be specific: prefer
billing.cancel_invoiceoverbilling.update. - Avoid generic names like
billing.executeorapi.call.
Each capability should map to a single, auditable action with clear side-effects.
Good:
billing.list_invoices(READ, no side-effects)billing.send_reminder(WRITE, sends an email)billing.void_invoice(DESTRUCTIVE, irreversible)
Avoid:
billing.do_stuff(too broad)billing.list_or_update_invoices(mixed safety classes)
| Class | Examples | Policy |
|---|---|---|
| READ | list, get, search, summarize | Always allowed |
| WRITE | create, update, send, approve | Justification + writer role |
| DESTRUCTIVE | delete, void, purge, terminate | Admin role only |
Use SensitivityTag.PII when results may contain: name, email, phone, SSN, address.
Use SensitivityTag.PCI when results may contain: card numbers, CVV, bank details.
Use SensitivityTag.SECRETS when results may contain: API keys, passwords, tokens.
Always pair sensitivity tags with allowed_fields to restrict which fields are returned
to non-privileged callers.
Add descriptive tags to improve keyword matching:
Capability(
capability_id="billing.list_invoices",
tags=["billing", "invoices", "list", "finance", "accounts receivable"],
...
)Kernel.invoke(..., dry_run=True) verifies the token and resolves the route
plan but never calls the driver. Use it to validate that a principal can
invoke a capability, inspect what a driver would receive, or run policy
checks in CI without live tool backends.
result = await kernel.invoke(
token,
principal=principal,
args={"operation": "billing.list_invoices", "max_rows": 5},
response_mode="summary",
dry_run=True,
)
# result: DryRunResult(
# capability_id="billing.list_invoices",
# principal_id="user-001",
# policy_decision=PolicyDecision(allowed=True, ...),
# driver_id="billing",
# operation="billing.list_invoices",
# resolved_args={"operation": "billing.list_invoices", "max_rows": 5},
# response_mode="summary",
# budget_remaining=None,
# estimated_cost="low",
# )Three rules govern dry-run behaviour — keep them in sync with the real-invoke path if you change either:
- Token verification still runs. Expired, revoked, or scope-mismatched
tokens raise
TokenExpired/TokenRevoked/TokenInvalid/TokenScopeErrorexactly as they would at real-invoke. Policy is not re-evaluated at invoke time — the granting policy decision is encoded in the token atgrant_capability. - Operation resolution mirrors drivers.
DryRunResult.operationis computed the same way every driver computes it:str(args.get("operation", capability_id)). Always useargs["operation"]when you need a fixed operation; otherwise the dry-run operation is the capability ID, matching what the driver would see. - Raw-mode admin gate mirrors the Firewall. Non-admin principals never
get
response_mode="raw"at real-invoke (the Firewall downgrades it to"summary"— seefirewall/transform.py). Dry-run downgrades the same way, so non-admin callers cannot probe for raw-mode availability viaDryRunResult.
The driver's execute() is never called in dry-run, so the mode is free of
side effects regardless of driver type (InMemoryDriver, HTTPDriver,
MCPDriver). DryRunResult.budget_remaining is currently always None; the
field is reserved for a future cross-invocation budget mechanism.
DeclarativePolicyEngine is an alternative to DefaultPolicyEngine that
loads rules from a YAML or TOML file (or a plain dict). Rules are evaluated
top-down, first-match-wins; if no rule matches, the policy's default action
applies ("deny" unless overridden).
from pathlib import Path
from agent_kernel import DeclarativePolicyEngine, Kernel
# YAML or TOML — both formats are interchangeable.
policy = DeclarativePolicyEngine.from_yaml(Path("examples/policies/default.yaml"))
# Or build entirely in-memory:
policy = DeclarativePolicyEngine.from_dict({
"default": "deny",
"rules": [
{"name": "allow-read", "action": "allow",
"match": {"safety_class": ["READ"], "sensitivity": ["NONE"]}},
# ...
],
})
kernel = Kernel(registry=registry, policy=policy)A rule's match block supports safety_class, sensitivity, roles
(ANY-of), attributes (ALL-of, with "*" meaning "attribute must be
present"), and min_justification (minimum stripped length). On allow, the
rule's constraints are merged into the resulting PolicyDecision. On
deny, reason is embedded in the raised PolicyDenied.
The DSL has no negation/missing-attribute operator today, so a policy that
should deny "when an attribute is missing" should be expressed as an allow
rule requiring the attribute paired with default: deny. See
examples/policies/default.yaml for a
worked example.
pyyaml and tomli are optional — they live behind the [policy]
extra. import agent_kernel always works; calling from_yaml / from_toml
without the parser installed raises PolicyConfigError with an install hint.
When a capability call is denied, Kernel.explain_denial(request, principal, justification="") returns a structured DenialExplanation describing
every unmet condition (not just the first one), so the caller can see the
full remediation path:
explanation = kernel.explain_denial(
CapabilityRequest(capability_id="billing.update_invoice", goal="..."),
principal,
justification="too short",
)
# explanation.denied == True
# explanation.rule_name == "write-min_justification"
# explanation.failed_conditions == [FailedCondition(condition="roles", required=[...]), ...]
# explanation.remediation == ["Add 'writer' or 'admin' role to ...", "Provide ..."]
# explanation.narrative == "Request for 'billing.update_invoice' by '...' would be denied: ..."Both built-in engines support explain(). If you bring a custom policy
engine that implements only PolicyEngine.evaluate, explain_denial raises
AgentKernelError with guidance — implement the ExplainingPolicyEngine
protocol to enable structured explanations.