All 33 files here live in
examples/ in the repository and are exercised
by the test suite — what you see below is exactly what’s on disk, not a
simplified retelling. Run any of them yourself with sense run examples/<name>.sns (see Installation if sense isn’t
on your PATH yet). A few near the end need a real ANTHROPIC_API_KEY
to do anything interesting — those check for the key first and print a
skip message if it’s missing, so they’re always safe to run.Getting started
hello.sns — the smallest Sense program
# The smallest Sense program.
print("hello, sense")
fib.sns — functions, recursion, typed return
# Deterministic core: functions, recursion, typed params/return.
def fib(n) -> Int:
if n < 2:
return n
return fib(n - 1) + fib(n - 2)
i = 0
while i < 10:
print(fib(i))
i = i + 1
arrays_and_loops.sns — arrays, for/in, break/continue
# Arrays, for-in, break/continue, builtins.
users: Array<String> = ["alice", "bob", "carol", "dave"]
for user in users:
if user == "carol":
continue
if user == "dave":
break
print("notifying " + user)
total = 0
for n in range(1, 6):
total = total + n
print("sum 1..5 =", total)
scores = [1, 2, 3]
push(scores, 4)
scores[0] = 10
print(scores)
modules/ — splitting a program across files
modules/math_utils.sns
# A module: importers see its top-level bindings as members.
def square(x) -> Int:
return x * x
def add(a, b) -> Int:
return a + b
pi = 3.14159
modules/main.sns
import "./math_utils.sns" as math
print(math.square(5))
print(math.add(2, 3))
print(math.pi)
sense run examples/modules/main.sns
25
5
3.14159
python_interop.sns — reaching into any installed Python package
# `import python "module"` reaches straight into any installed Python
# package. Tractable because Sense's own interpreter already runs inside
# the same Python process -- no FFI, no serialization, just importlib +
# getattr underneath.
import python "math" as math
print(math.sqrt(16))
print(math.pi)
# Most Sense values already ARE Python values, so they cross the boundary
# for free -- a Python list that comes back is just a Sense Array, usable
# with the native array builtins immediately:
import python "json" as json
data = json.loads("[1, 2, 3]")
push(data, 4)
print(data)
print(json.dumps(data))
# A dict has no native Sense type, so it comes back as an opaque wrapper --
# still usable via indexing and member access, just not fully "a Sense
# value":
person = json.loads("{\"name\": \"ada\", \"age\": 36}")
print(person["name"])
person["age"] = 37
print(person["age"])
# The safety model still applies -- but only to what you wrap in an
# action. A bare import + call is exactly as unrestricted as a plain
# Sense fn (that's an existing boundary, not a new hole):
irreversible action compute_root(x: Int) requires math.compute:
return math.sqrt(x)
policy: allow math.compute
result = compute_root(25)
result.verify()
print(result.commit())
testing.sns — the built-in testing framework
# Phase 5: a real testing framework. Run this with `sense test
# examples/testing.sns` (not `sense run`) to see the pass/fail report --
# though `sense run` executes it too, since a test file is just a normal
# Sense program with some `test` blocks in it.
def square(x) -> Int:
return x * x
def fib(n) -> Int:
if n < 2:
return n
return fib(n - 1) + fib(n - 2)
test "square of a positive number":
assert(square(3) == 9)
test "square of zero":
assert(square(0) == 0, "square(0) should be 0")
test "fibonacci base cases":
assert(fib(0) == 0)
assert(fib(1) == 1)
test "fibonacci recursive case":
assert(fib(6) == 8)
# A `test` block's failure is isolated -- it's recorded, not raised, so it
# does not stop the rest of the file or the rest of the suite from running.
test "this one is deliberately wrong":
assert(square(2) == 5, "2 squared is 4, not 5 -- this failure is expected")
print("the file itself finished running top to bottom")
docstrings_basics.sns — .doc on functions, tools, actions, and agents
def square(x: Int) -> Int:
"returns the square of x"
return x * x
tool greet(name: String) returns String:
"""
Greets a user by name.
"""
return "hi " + name
reversible action rename_file(old_path: String, new_path: String) requires fs.rename:
"""
Renames a file on disk. Reversible: rollback renames it back.
"""
print("renaming " + old_path + " -> " + new_path)
rollback:
print("reverting: renaming " + new_path + " -> " + old_path)
agent researcher:
"Investigates a topic and records what it finds as a plain variable."
finding = "sense uses indentation, not braces"
policy: allow fs.rename
print(square(6))
print(greet("ada"))
fix = rename_file("draft.txt", "final.txt")
fix.verify()
fix.commit()
fix.rollback()
start(researcher)
print(researcher.doc)
print(researcher.finding)
36
hi ada
renaming draft.txt -> final.txt
reverting: renaming final.txt -> draft.txt
Investigates a topic and records what it finds as a plain variable.
sense uses indentation, not braces
.doc — nothing new to write, that string was already
legal there. sense inspect shows it in every panel.
See also: Functions: Docstrings
Safety and control
actions_and_policy.sns — prepare → verify → commit, and capability policy
# Phase 4: Action (prepare -> verify -> commit) and policy.
#
# Central law: preparing an action must never cause its external effect.
# Calling send_mail(...) below does NOT send anything -- only .commit()
# does.
irreversible action send_mail(to: String, body: String) requires email.send:
print("SENDING to " + to + ": " + body)
policy: allow email.send
mail = send_mail("alice@example.com", "the meeting moved to 3pm")
print("state after prepare: " + mail.state) # "prepared" -- nothing sent yet
mail.verify()
print("state after verify: " + mail.state) # "verified"
mail.commit() # only now does the effect happen
print("state after commit: " + mail.state) # "committed"
# A denied capability blocks .verify() (and, independently, .commit() too --
# verification is not a permanent guarantee, commit re-checks).
irreversible action delete_account(user: String) requires account.delete:
print("DELETING " + user)
policy: deny account.delete
risky = delete_account("bob")
# risky.verify() # would raise SensePolicyError: denied by policy
# policy scopes like delegation does -- a session or agent can grant itself
# a capability without changing the outer policy.
session checkout:
policy: allow account.delete
cleanup = delete_account("temp-user")
cleanup.verify()
cleanup.commit()
# outside the session, account.delete is denied again:
still_denied = delete_account("bob")
# still_denied.verify() # would still raise
action_rollback.sns — rollback: and action.rollback()
# `rollback:` / `action.rollback()` -- reversible-action compensation.
# A `reversible action` must declare `rollback:`; an `irreversible action`
# may not, since it "cannot be reliably undone."
reversible action update_profile(user: String, bio: String) requires profile.write:
print("SETTING " + user + "'s bio to: " + bio)
rollback:
print("REVERTING " + user + "'s bio")
policy: allow profile.write
a = update_profile("alice", "new bio")
a.verify()
a.commit()
print("state after commit: " + a.state)
a.rollback()
print("state after rollback: " + a.state)
# .rollback() re-checks the capability at rollback time -- same "never trust
# stale verification" principle .commit() already applies. A denial here
# leaves the action's state at "committed": nothing changed, so nothing
# should look changed.
policy: deny profile.write
# a.rollback() # would raise SensePolicyError -- already rolled back anyway
reversible_db_register.sns — a reversible action that isn’t a file
import python "sqlite3" as sqlite3
db = sqlite3.connect("users_demo.db")
db.execute("CREATE TABLE IF NOT EXISTS users (email TEXT PRIMARY KEY, plan TEXT)")
db.commit()
reversible action register_user(email: String, plan: String) requires user.register:
db.execute("INSERT INTO users (email, plan) VALUES (?, ?)", [email, plan])
db.commit()
print("registered " + email + " on the " + plan + " plan")
rollback:
db.execute("DELETE FROM users WHERE email = ?", [email])
db.commit()
print("deprovisioned " + email)
policy:
allow user.register
signup = register_user("newuser@example.com", "pro")
signup.verify()
signup.commit()
row_count = db.execute("SELECT COUNT(*) FROM users WHERE email = ?", ["newuser@example.com"]).fetchone()[0]
print("rows right after commit: " + str(row_count))
signup.rollback()
row_count = db.execute("SELECT COUNT(*) FROM users WHERE email = ?", ["newuser@example.com"]).fetchone()[0]
print("rows after rollback: " + str(row_count))
db.close()
registered newuser@example.com on the pro plan
rows right after commit: 1
deprovisioned newuser@example.com
rows after rollback: 0
rollback: really deletes the row it just inserted.
See also: Rollback ·
Python Interop
dry_run_basics.sns — preview what would commit, without letting it
reversible action send_email(to: String, subject: String) requires email.send:
print(" -> actually sending to " + to)
rollback:
print(" -> recalling email to " + to)
irreversible action delete_account(user: String) requires account.delete:
print(" -> actually deleting " + user)
policy:
allow email.send
allow account.delete
recipients = ["alice@example.com", "bob@example.com", "carol@example.com"]
for person in recipients:
mail = send_email(person, "Your weekly digest")
mail.verify()
mail.commit()
cleanup = delete_account("inactive_user@example.com")
cleanup.verify()
cleanup.commit()
sense run --dry-run examples/dry_run_basics.sns
DRY RUN — no actions will actually commit
...
--- dry run summary ---
[line 1] reversible action 'send_email' -> prepared
[line 2] reversible action 'send_email' -> verified
reversible action 'send_email' -> committed (dry run — body not executed)
... (one more per recipient)
irreversible action 'delete_account' -> prepared
irreversible action 'delete_account' -> verified
irreversible action 'delete_account' -> committed (dry run — body not executed)
--dry-run and every "-> actually ..." line prints for
real. With it, those lines vanish completely — every capability check
still runs for real, only the body that performs the effect is skipped.
See also: Actions: Dry run ·
CLI Reference
real_email_action.sns — a real side effect, gated and previewable
import python "smtplib" as smtplib
SMTP_HOST: String = "localhost"
SMTP_PORT: Int = 25
SMTP_FROM: String = "sense@example.com"
SMTP_TO: String = "test@example.com"
irreversible action send_real_email(to: String, subject: String, body: String) requires email.send:
message = "From: " + SMTP_FROM + "\r\nTo: " + to + "\r\nSubject: " + subject + "\r\n\r\n" + body
server = smtplib.SMTP(SMTP_HOST, SMTP_PORT)
server.sendmail(SMTP_FROM, [to], message)
server.quit()
print("sent to " + to + " via " + SMTP_HOST + ":" + str(SMTP_PORT))
policy:
allow email.send
mail = send_real_email(SMTP_TO, "Hello from Sense", "really sent by smtplib, caught by smtp4dev")
mail.verify()
mail.commit()
email.send is irreversible on purpose, since
an email can’t be unsent. Preview it with no smtp4dev running at all:
sense run --dry-run examples/real_email_action.sns
DRY RUN — no actions will actually commit
state: committed
--- dry run summary ---
[line 1] irreversible action 'send_real_email' -> prepared
[line 2] irreversible action 'send_real_email' -> verified
irreversible action 'send_real_email' -> committed (dry run — body not executed)
approval_and_audit.sns — mandatory approval and the audit log
# A declared "requires approval" marker on an action, composed with
# ask_human (see examples/ask_human.sns), plus the audit log that now
# records every prepare/verify/commit/deny/fail event.
irreversible action delete_account(user: String) requires account.delete, approval:
print("DELETING " + user)
policy: allow account.delete
a = delete_account("bob")
a.verify()
print("approved before asking: " + str(a.approved)) # false
# The action mandates approval -- committing without it is an error, not
# just something the caller forgot to add manually:
# a.commit() # would raise SenseApprovalError
agent reviewer:
answer = ask_human("approve deleting bob? (yes/no)")
if answer == "yes":
a.approve()
start(reviewer)
resume(reviewer, "yes")
print("approved after asking: " + str(a.approved)) # true
a.commit()
print("---- audit log ----")
for line in audit_log():
print(line)
policy_library.sns + policy_true_potential.sns — one library, three trust levels
A shared capability library declares no policy of its own — what’s
actually allowed is entirely up to whoever imports it:
policy_library.sns
tool refund_customer(order_id: String, amount: Int) returns String requires payment.refund "refund a customer for a specific order":
return "refunded $" + str(amount) + " for order " + order_id
tool send_receipt(email: String, order_id: String) returns String requires email.send "email a receipt for an order":
return "receipt sent to " + email + " for order " + order_id
irreversible action delete_customer_data(customer_id: String) requires data.delete:
print("PERMANENTLY deleting all data for " + customer_id)
policy_true_potential.sns
import "./policy_library.sns" as lib
session first_line_support:
policy:
deny payment.refund
allow email.send
receipt = lib.send_receipt("customer@example.com", "ORD-104")
print("[first line] " + receipt)
# lib.refund_customer("ORD-104", 40) # would raise SensePolicyError
session senior_support:
policy:
allow payment.refund
allow email.send
refund = lib.refund_customer("ORD-104", 40)
print("[senior] " + refund)
session incident_response:
policy:
allow data.delete
cleanup = lib.delete_customer_data("cust-9981")
cleanup.verify()
cleanup.commit()
[first line] receipt sent to customer@example.com for order ORD-104
[senior] refunded $40 for order ORD-104
PERMANENTLY deleting all data for cust-9981
sessions, three different sets
of permissions — nothing about policy_library.sns changed between
them. A capability check resolves against the caller’s scope, not the
declaration’s, so this is real gating, not documentation.
See also: Policy ·
Variables & Scope
Agents, sessions, and pausing
session_and_agent.sns — a scope vs. a runtime entity
# Phase 3: session (scoped execution/context boundary) and agent (a real
# runtime entity: identity, persistent state, a partial lifecycle).
model outer_model = inference("mock", "outer", response_template: "outer says: {prompt}")
set delegation = outer_model
# A session is NOT an autonomous actor -- it's just a scope. Delegation set
# inside it does not leak back out once the session ends.
session researcher:
model inner_model = inference("mock", "inner", response_template: "inner says: {prompt}")
set delegation = inner_model
print(ask("best database?").value)
print(ask("still outer?").value)
# An agent IS a runtime entity: declaring it only reaches "created" -- its
# body does not run until start(agent) is called.
agent analyst:
goal: String = "summarize the findings"
set delegation = outer_model
finding = ask(goal)
print("analyst: " + finding.value)
print(analyst.status)
start(analyst)
print(analyst.status)
print(analyst.goal)
# Starting an already-completed agent is an error -- there's no pause/
# resume or re-run yet.
# start(analyst) # would raise: already been started
agent_pause_resume.sns — suspending and resuming mid-body
# Agent pause/resume: an agent can suspend itself mid-body (e.g. to wait for
# human approval) and be resumed later, continuing exactly where it left
# off -- local variables and all.
agent approver:
print("checking the request")
amount = 5000
pause("needs approval for amount over 1000: " + str(amount))
print("approved -- proceeding")
print("charging " + str(amount))
print(approver.status) # "created" -- nothing has run yet
start(approver)
print(approver.status) # "paused"
print(approver.pause_reason)
# ... elsewhere, e.g. a human reviews and approves ...
resume(approver)
print(approver.status) # "completed"
ask_human.sns — human-in-the-loop, built on pause/resume
# Human-in-the-loop, built on pause()/resume(): ask_human(question) suspends
# a running agent exactly like pause(), but the human's answer -- whatever
# resume(agent, answer) provides -- becomes the call's return value.
agent approver:
amount = 5000
answer = ask_human("approve charge of " + str(amount) + "? (yes/no)")
if answer == "yes":
print("charging " + str(amount))
else:
print("cancelled")
start(approver)
print(approver.status) # "paused"
print(approver.pause_reason) # the question
# ... elsewhere, e.g. a human reviews and answers ...
resume(approver, "yes")
print(approver.status) # "completed"
# Outside any agent, ask_human() has no one to hand control back to, so it
# falls back to reading a real line from stdin instead of suspending:
#
# name = ask_human("what is your name?")
# print("hello, " + name)
async_basics.sns — opt-in concurrency for agent and tool
# `async agent` / `async tool` -- concurrency is opt-in at this boundary,
# not threaded through every plain function call. A plain `agent`/`tool`
# (no `async`) behaves exactly as before: `start()`/calling a tool blocks
# until it's done. `async` skips that blocking wait.
import python "time" as time
async agent fetch_a:
time.sleep(0.1)
result = "data from A"
async agent fetch_b:
time.sleep(0.1)
result = "data from B"
start(fetch_a) # returns immediately -- fetch_a.status is "running"
start(fetch_b) # also returns immediately -- both bodies now run concurrently
await(fetch_a) # blocks until fetch_a settles
await(fetch_b) # blocks until fetch_b settles
print(fetch_a.result)
print(fetch_b.result)
# An async tool call returns a Future immediately; await() blocks for the
# result. Capability checks (if `requires` is declared) still happen
# synchronously at the call site -- a denial doesn't need await() to surface.
async tool slow_search(query: String) returns String:
time.sleep(0.05)
return "results for " + query
f = slow_search("sense")
print(type_of(f)) # "Future"
print(f.done) # probably false -- the body is still sleeping
print(await(f)) # blocks until the sleep finishes, then returns the result
print(f.done) # true, now that it's been awaited
agent_communication.sns — send/receive between concurrent agents
async agent worker_a:
task = receive()
print("worker_a: researching '" + task + "'")
send(coordinator, "worker_a found: population data for " + task)
async agent worker_b:
task = receive()
print("worker_b: researching '" + task + "'")
send(coordinator, "worker_b found: climate data for " + task)
async agent coordinator:
send(worker_a, "Iceland")
send(worker_b, "Iceland")
first = receive()
second = receive()
print("coordinator: " + first)
print("coordinator: " + second)
print("coordinator: both workers reported in")
start(worker_a)
start(worker_b)
start(coordinator)
await(coordinator)
await(worker_a)
await(worker_b)
worker_a: researching 'Iceland'
worker_b: researching 'Iceland'
coordinator: worker_a found: population data for Iceland
coordinator: worker_b found: climate data for Iceland
coordinator: both workers reported in
send() never
blocks; receive() blocks that agent’s own thread until something
arrives. Sharing a plain variable between two async agents instead
would race — this is the safe way to hand data between them.
See also: Agents: Communication
Memory
memory_basics.sns — a session-scoped key/value store
# `memory` -- a persistent-for-the-run key/value store, distinct from an
# ordinary variable.
#
# `: "kind"` is a free-form, unvalidated hint (short-term/long-term/episodic/
# semantic/working are *potential* categories, not a closed set) kept only
# for introspection.
memory notes: "working"
notes.remember("topic", "sense keywords")
notes.remember("status", "in progress")
print(notes.recall("topic"))
print(notes.recall("missing")) # nil -- unlike a bare variable, no NameError
print(notes.keys())
notes.forget("status")
print(notes.keys())
print(notes.kind)
# Backed by an in-process dict -- explicitly not persisted across runs.
# Real retrieval/indexing/expiration stays future work.
persistent_memory_basics.sns — durable, versioned storage
# `persistent memory` -- durable, versioned, append-only key/value storage,
# scoped down from a literal Delta Lake to the part worth keeping: an
# append-only log gives versioning and auditing as the same mechanism.
# Backed by a SQLite file next to wherever this runs -- deleted first
# below so re-running this example starts clean each time.
import python "os" as os
if os.path.exists("ledger.sense_memory.db"):
os.remove("ledger.sense_memory.db")
persistent memory ledger
ledger.remember("name", "ada")
print(ledger.recall("name")) # "ada"
ledger.remember("name", "ada lovelace")
print(ledger.recall("name")) # "ada lovelace" -- the latest value
# remember() never overwrites in place -- it's a new row every time, so
# nothing here is ever lost, just superseded.
for line in ledger.history("name"):
print(line)
# Time travel: the value as of an earlier version.
print(ledger.as_of("name", 1)) # "ada" -- what it was before the second remember()
ledger.remember("status", "draft")
print(ledger.keys()) # [name, status]
ledger.forget("status")
print(ledger.recall("status")) # nil
print(ledger.keys()) # [name] -- forgotten keys drop out of keys()
print(len(ledger.history("status"))) # 2 -- but the forget is still on record
print(ledger.path)
# Durability: a brand-new Interpreter pointed at the same file sees this
# state -- that's the actual feature. (Demonstrated in
# tests/test_persistent_memory.py::test_durable_across_separate_interpreter_instances
# by literally constructing a second Interpreter; a single Sense program
# can't observe "a different process" from inside itself.)
tool_basics.sns — capability-checked functions
# `tool` -- a capability-checked function.
#
# Distinct from `action`: calling a tool runs its body immediately, no
# prepare -> verify -> commit lifecycle. It exists for things that aren't
# necessarily world-changing (a search, a lookup) but should still be
# typed, capability-gated, and audited -- unlike a bare `fn`.
tool search(query: String) returns String requires web.search:
return "results for " + query
policy: allow web.search
print(search("sense lang"))
# A tool with no `requires` clause is just a documented, typed interface --
# unrestricted, same as a plain `fn`.
tool add(a: Int, b: Int) returns Int:
return a + b
print(add(2, 3))
# Every tool call is recorded in the same audit log actions use.
for line in audit_log():
print(line)
Real LLM integration
Every example in this section checks for
ANTHROPIC_API_KEY first and
prints a skip message if it’s not set — safe to run without a key, but
you’ll only see the interesting output with one configured. See
Installation for the [anthropic]/[mcp] optional
extras these need.ask_basics.sns — model / ask / answer, fully offline
# Phase 2: model / ask / answer.
#
# `ask(...)` never talks to a vendor API directly -- it asks whichever
# model is currently delegated to. Here we delegate to an offline mock model
# so this example needs no network access or API key.
#
# There is no `model` keyword: inference(...) already returns a Model
# value, so an ordinary assignment is enough -- the runtime type carries
# the meaning, not a declaration keyword.
model reasoning = inference("mock", "demo-mock", response_template: "I think the answer relates to: {prompt}")
set delegation = reasoning
result = ask("what database should we use?")
print(type_of(result))
print(result)
print(result.value)
print(result.confidence)
print(result.source)
# A typed answer binding -- `answer<T>` is spelled lowercase in an
# annotation, same as every other type, narrowed to primitive payload
# types for now (no structs yet).
country: answer<String> = ask("is the sky blue?")
print(country.value)
# Explicit form: call .ask() directly on a specific model, bypassing
# delegation entirely.
model backup = inference("mock", "backup-mock")
backup_result = backup.ask("plan B")
print(backup_result)
real_model_basics.sns — a real provider, inference("anthropic", ...)
# A real model provider: `inference("anthropic", ...)`.
#
# Same `ask()`/`Answer` surface as ask_basics.sns -- the runtime decides
# how reasoning is implemented, so switching from inference("mock", ...)
# to a real provider needs no other code change.
# `inference(...)` is the one function for every real provider (see
# real_model_openai_basics.sns for the identical shape with "openai"
# instead) -- not a separate `anthropic_model(...)` to learn.
#
# Requires a real ANTHROPIC_API_KEY (this makes a real, costed network
# call) -- this example checks for it first and skips gracefully if it's
# not set, so it stays runnable (and free) in CI/offline.
import python "os" as os
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("skip: set ANTHROPIC_API_KEY to run this example")
else:
# `model claude = ...` -- the declaration names it; inference() itself
# takes no separate name argument. temperature:/max_tokens: are
# optional labeled arguments, forwarded to the real API call (not
# just accepted and ignored) -- omit either to use the provider's own
# default.
model claude = inference("anthropic", "claude-sonnet-5", temperature: 0.3, max_tokens: 300)
# A real call costs money and sends prompt data to a third party --
# that's exactly why it's capability-gated, unlike inference("mock", ...).
# This policy is optional: with no policy at all, the call is allowed by
# default, same as `tool`/`action`.
policy: allow model.anthropic
set delegation = claude
result = ask("in one short sentence, why might someone want an AI-native programming language?")
print(result.value)
print(result.confidence) # nil -- no fabricated confidence, unlike inference("mock", ...)
for line in audit_log():
print(line) # the real call is recorded, unlike a mock one
real_model_openai_basics.sns — the same shape, a second vendor
import python "os" as os
if os.environ.get("OPENAI_API_KEY") == nil:
print("skip: set OPENAI_API_KEY to run this example")
else:
model gpt = inference("openai", "gpt-4o", temperature: 0.3, max_tokens: 300)
policy: allow model.openai
set delegation = gpt
result = ask("in one short sentence, why might someone want an AI-native programming language?")
print(result.value)
print(result.confidence) # nil -- no fabricated confidence, unlike inference("mock", ...)
real_model_basics.sns with "anthropic" swapped for
"openai" — nothing else changes. One function for every real provider,
not a separate one to learn per vendor.
See also: Model, Ask & Answer
ask_with_tools_basics.sns — a model choosing which tool to call
# `ask_with_tools(prompt, tools)` -- a real model choosing which `tool`
# to invoke. The model proposes; `_call_tool` still disposes: a tool the
# model asks for goes through the exact same capability/type-check/audit
# path a hand-written call reaches, so a policy-denied tool is never
# executed just because a model asked for it.
#
# Requires a real ANTHROPIC_API_KEY (this makes real, costed network
# calls) -- this example checks for it first and skips gracefully if it's
# not set, so it stays runnable (and free) in CI/offline.
import python "os" as os
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("skip: set ANTHROPIC_API_KEY to run this example")
else:
tool get_weather(city: String) returns String requires weather.read "get the current weather for a city":
return "sunny, 22C"
tool charge_card(amount: Int) returns String requires payment.execute "charge a customer's card":
return "charged " + str(amount)
policy:
allow weather.read
deny payment.execute
model claude = inference("anthropic")
set delegation = claude
# The model isn't told which tool to call -- it reads the prompt and
# each tool's description, and decides.
weather = ask_with_tools("what's the weather like in Paris right now?", [get_weather, charge_card])
print(weather.value)
# Ask for something that would require the denied tool. The model may
# attempt to call charge_card -- if it does, the call never runs; the
# denial is fed back to the model (which responds accordingly) and
# recorded in the audit log, exactly like a hand-written denied call.
billing = ask_with_tools("please charge my card $50 for the subscription", [get_weather, charge_card])
print(billing.value)
print("--- audit log ---")
for line in audit_log():
print(line)
memory_wired_into_reasoning.sns — retrieval and write-back
# Memory wired into reasoning: retrieval-into-prompt + write-back.
#
# Retrieval has one small, new builtin: ask_with_memory(prompt, memory)
# injects the memory's entire current contents as context ahead of the
# prompt -- works with inference("mock", ...) too, since it's just string
# composition ahead of an ordinary ask() call.
#
# Write-back needed no new interpreter feature at all: it already composes
# out of `memory` + a small `tool` wrapper + ask_with_tools -- the
# model decides what's worth remembering, capability-gated and audited
# exactly like any other tool call.
memory profile
profile.remember("favorite_color", "blue")
model demo_model = inference("mock", "demo", response_template: "Given what you told me, {prompt}")
set delegation = demo_model
# -- retrieval: the model sees the memory's contents without any tool call
answer = ask_with_memory("what is my favorite color?", profile)
print(answer.value)
# -- write-back: only meaningful with a real, tool-calling-capable model,
# so this half is guarded the same way real_model_basics.sns is.
import python "os" as os
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("skip: set ANTHROPIC_API_KEY to run the write-back half of this example")
else:
tool remember_fact(key: String, value: String) returns String "remember a fact about the user for later":
profile.remember(key, value)
return "ok"
tool recall_fact(key: String) returns String "recall a previously remembered fact about the user":
result = profile.recall(key)
if result == nil:
return "(unknown)"
return result
model claude = inference("anthropic")
set delegation = claude
prompt = "remember that my favorite food is sushi, then tell me my favorite color and favorite food"
result = ask_with_tools(prompt, [remember_fact, recall_fact])
print(result.value)
# Genuinely persisted -- readable back without any model involved.
print(profile.recall("favorite_food"))
skills_basics.sns — bundling tools under one description
# `skill(name, description, tools)` -- a named, reusable bundle of `tool`s.
#
# No dedicated keyword, same reasoning as why there's no `model` keyword:
# the builtin's return type (a `Skill` value) already carries the meaning
# wherever it's checked, here in ask_with_tools(). The description
# isn't just documentation -- it's prepended to the prompt as an
# "Available skills:" preamble, so it has real, not cosmetic, effect.
#
# Requires a real ANTHROPIC_API_KEY (this makes real, costed network
# calls) -- this example checks for it first and skips gracefully if it's
# not set, so it stays runnable (and free) in CI/offline.
import python "os" as os
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("skip: set ANTHROPIC_API_KEY to run this example")
else:
tool search(query: String) returns String requires web.search "search the web for a query":
return "results for " + query
tool fetch_page(url: String) returns String requires web.fetch "fetch the contents of a web page":
return "contents of " + url
tool send_email(to: String, body: String) returns String requires email.send "send an email":
return "sent to " + to
policy:
allow web.search
allow web.fetch
allow email.send
web_research = skill("web_research", "search the web and fetch pages to answer questions", [search, fetch_page])
email_skill = skill("email", "compose and send emails on the user's behalf", [send_email])
model claude = inference("anthropic")
set delegation = claude
# The model sees both skills' descriptions (not just their tools) and
# decides which one applies -- reusable across as many calls as you
# like, instead of re-listing individual tools each time.
result = ask_with_tools("what's the latest news about AI-native programming languages?", [web_research, email_skill])
print(result.value)
for line in audit_log():
print(line)
mcp_client_basics.sns — tools from an external MCP server
# `connect_mcp(name, command, args, capability?)` -- connecting to an
# external MCP server for tools a real model can invoke. The genuinely
# new piece here: an MCP tool has no Sense body at all -- invoking one is
# a JSON-RPC call to an external process, not local Sense code.
#
# Demonstrated against this repo's own tiny test MCP server
# (tests/mcp_test_server.py, a real server, not a mock) so this example
# is runnable without Node.js/npx -- in real usage you'd point `command`/
# `args` at a real MCP server package instead (many are `npx`-based).
#
# Requires a real ANTHROPIC_API_KEY (this makes real, costed network
# calls) and the optional `mcp` package (`pip install sense-lang[mcp]`).
# Sense has no exception handling, so there's no graceful way to detect
# "the mcp package isn't installed" from inside a Sense program -- this
# example can only guard on the one thing it *can* check before calling
# anything: the API key. Run it with a key set but without `mcp`
# installed, and connect_mcp() raises a plain, uncaught SenseRuntimeError
# -- the same limitation `inference("anthropic", ...)` already has for
# its own optional dependency.
import python "os" as os
import python "sys" as sys
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("skip: set ANTHROPIC_API_KEY to run this example")
else:
server = connect_mcp("calc", sys.executable, ["tests/mcp_test_server.py"], "mcp.calc")
policy: allow mcp.calc
model claude = inference("anthropic")
set delegation = claude
# The model sees this real server's own tool schemas and descriptions
# -- discovered at connect_mcp() time, not written anywhere in this
# Sense source -- and decides whether/how to call them.
result = ask_with_tools("what is 17 plus 25?", server.tools)
print(result.value)
for line in audit_log():
print(line)
server.close()
Putting it together
incident_response/ — an SRE copilot, every feature for a real reason
One coherent scenario, not a checklist: three incidents come off a
monitoring feed; for each, two diagnostics run concurrently and report to
a coordinator; then, based on severity, the incident is auto-remediated
(low) or escalated to a human (high) — enforced by a session-scoped
policy, not just an if/else, so a branching-logic bug still can’t
auto-remediate a high-severity incident.
incident_response/runbook_library.sns
memory service_status: "simulated live health, keyed by service name"
service_status.remember("checkout-api", "degraded")
service_status.remember("payments-api", "degraded")
service_status.remember("auth-service", "disk_full")
tool check_service_health(service: String) returns String requires monitoring.read "check a service's current live health status":
status = service_status.recall(service)
if status == nil:
return "unknown"
if service == "payments-api":
return "degraded" # a restart alone never actually fixes this one
return status
tool lookup_runbook(issue: String) returns String requires monitoring.read "look up the standard remediation runbook for a known issue type":
if issue == "high_latency" or issue == "degraded":
return "restart the service process; re-check health afterward"
if issue == "disk_full":
return "do not auto-remediate -- disk-full needs a human to decide what's safe to delete; escalate"
return "no known runbook for this issue -- escalate"
reversible action restart_service(service: String) requires ops.remediate:
print(" -> restarting " + service)
service_status.remember(service, "healthy")
rollback:
print(" -> restart didn't actually clear it -- reverting " + service + "'s recorded status")
service_status.remember(service, "degraded")
irreversible action page_oncall(engineer: String, message: String) requires paging.send, approval:
print(" -> paging " + engineer + ": " + message)
incident_response/main.sns
import python "os" as os
import python "json" as json
import "./runbook_library.sns" as ops
if os.path.exists("incident_ledger.sense_memory.db"):
os.remove("incident_ledger.sense_memory.db")
persistent memory incident_ledger: "incident_ledger.sense_memory.db"
memory triage_notes: "working"
def is_auto_remediable(severity: String) -> Bool:
return severity == "low"
test "low severity is auto-remediable":
assert(is_auto_remediable("low") == true)
test "high severity requires escalation, not auto-remediation":
assert(is_auto_remediable("high") == false)
model triage_model = inference("mock", "triage-assistant", response_template: "triage note -- {prompt}")
set delegation = triage_model
# Advanced, optional: a real model choosing which diagnostic tool to run --
# needs a real ANTHROPIC_API_KEY, skipped gracefully without one.
if os.environ.get("ANTHROPIC_API_KEY") == nil:
print("(skipping the real-model tool-selection demo -- set ANTHROPIC_API_KEY to see it)")
else:
diagnostics = skill("diagnostics", "check a service's live health or look up its runbook", [ops.check_service_health, ops.lookup_runbook])
model claude = inference("anthropic")
set delegation = claude
triage_call = ask_with_tools("auth-service is reporting disk_full -- what do you recommend?", [diagnostics])
print("model recommendation: " + triage_call.value)
set delegation = triage_model
raw_incidents = "[{\"service\": \"checkout-api\", \"severity\": \"low\", \"issue\": \"high_latency\", \"oncall\": \"none\"}, {\"service\": \"payments-api\", \"severity\": \"low\", \"issue\": \"high_latency\", \"oncall\": \"none\"}, {\"service\": \"auth-service\", \"severity\": \"high\", \"issue\": \"disk_full\", \"oncall\": \"priya\"}]"
incidents = json.loads(raw_incidents)
for incident in incidents:
service = incident["service"]
severity = incident["severity"]
issue = incident["issue"]
print("")
print("=== incident: " + service + " (" + severity + " / " + issue + ") ===")
note = ask_with_memory("what do we already know about " + service + "?", triage_notes)
print(note.value)
triage_notes.remember(service, issue)
session handle_incident:
# Fan-out/fan-in: two diagnostics run concurrently, report back to
# a coordinator over a mailbox -- not a shared variable, which
# would race.
async agent disk_investigator:
health = ops.check_service_health(service)
send(triage_coordinator, "health check: " + health)
async agent network_investigator:
runbook = ops.lookup_runbook(issue)
send(triage_coordinator, "runbook says: " + runbook)
async agent triage_coordinator:
finding_a = receive()
finding_b = receive()
print(" " + finding_a)
print(" " + finding_b)
start(disk_investigator)
start(network_investigator)
start(triage_coordinator)
# Remediation must not begin until both diagnostics report in.
await(triage_coordinator)
await(disk_investigator)
await(network_investigator)
incident_ledger.remember(service, "triaged: " + issue)
if is_auto_remediable(severity):
policy:
allow ops.remediate
deny paging.send
# Declared *inside* this branch on purpose: a plain function
# closes over the scope it was declared in, so its capability
# checks run against *this* policy, not the global default.
def attempt_remediation(target: String) -> String:
fix = ops.restart_service(target)
fix.verify()
fix.commit()
after = ops.check_service_health(target)
if after == "healthy":
return target + ": remediated automatically"
fix.rollback()
return target + ": restart didn't clear it, rolled back -- needs a human"
outcome = attempt_remediation(service)
print(" " + outcome)
incident_ledger.remember(service, outcome)
else:
policy:
deny ops.remediate
allow paging.send
# Human-in-the-loop: pauses on ask_human(), standing in for a
# real page/Slack message. resume() supplies a canned reply so
# this file stays runnable non-interactively.
agent incident_commander:
answer = ask_human("page " + incident["oncall"] + " for " + service + "? (yes/no)")
start(incident_commander)
print(" " + incident_commander.pause_reason)
resume(incident_commander, "yes")
if incident_commander.answer == "yes":
page = ops.page_oncall(incident["oncall"], severity + " severity " + issue + " on " + service)
page.approve()
page.verify()
page.commit()
incident_ledger.remember(service, "escalated to " + incident["oncall"])
else:
print(" escalation declined -- no page sent")
incident_ledger.remember(service, "escalation declined")
print("")
print("=== durable history for payments-api (the one that needed a rollback) ===")
for line in incident_ledger.history("payments-api"):
print(line)
print("")
print("=== full audit trail ===")
for line in audit_log():
print(line)
sense run examples/incident_response/main.sns
(skipping the real-model tool-selection demo -- set ANTHROPIC_API_KEY to see it)
=== incident: checkout-api (low / high_latency) ===
triage note -- what do we already know about checkout-api?
health check: degraded
runbook says: restart the service process; re-check health afterward
-> restarting checkout-api
checkout-api: remediated automatically
=== incident: payments-api (low / high_latency) ===
triage note -- Known information from memory:
- checkout-api: high_latency
what do we already know about payments-api?
health check: degraded
runbook says: restart the service process; re-check health afterward
-> restarting payments-api
-> restart didn't actually clear it -- reverting payments-api's recorded status
payments-api: restart didn't clear it, rolled back -- needs a human
=== incident: auth-service (high / disk_full) ===
triage note -- Known information from memory:
- checkout-api: high_latency
- payments-api: high_latency
what do we already know about auth-service?
health check: disk_full
runbook says: do not auto-remediate -- disk-full needs a human to decide what's safe to delete; escalate
page priya for auth-service? (yes/no)
-> paging priya: high severity disk_full on auth-service
=== durable history for payments-api (the one that needed a rollback) ===
[v3] remember 'payments-api' = triaged: high_latency (...)
[v4] remember 'payments-api' = payments-api: restart didn't clear it, rolled back -- needs a human (...)
=== full audit trail ===
[line 113] tool 'check_service_health' -> called (via agent:disk_investigator)
[line 117] tool 'lookup_runbook' -> called (via agent:network_investigator)
[line 160] reversible action 'restart_service' -> prepared
[line 161] reversible action 'restart_service' -> verified
reversible action 'restart_service' -> committed
[line 163] tool 'check_service_health' -> called
...
audit_log() printed at the end, with (via agent:...) showing exactly
which concurrent agent triggered which call.
See also: Actions · Agents ·
Policy · Memory
Continue
Keywords reference
Every keyword used above, with full syntax and when to reach for it.
Quickstart
A guided first program, if you’d rather start smaller than this page.

