Wire megan-tk into a real control loop — the change-doer pattern, rate limits, error handling, and composition.
This page is for going past “hello world” — integrating megan-tk into a real
system. The mental model: megan-tk decides, your change-doer acts. The token
never touches actuators; it hands you a verdict, and your code owns the change.
Your perception produces the state megan-tk needs — which milestones are
reached, what’s blocking, or raw position/goal/obstacles.
2
Govern (megan-tk)
step / perceive returns a Decision. This is the only thing megan-tk
does — it does not move the robot.
3
Change (you)
On adapt, your change-doer intervenes at frontier — re-plan, switch
skill, correct a trajectory. On continue, let the policy run.
from cadenza_cli import MeganTKtk = MeganTK()with tk.session("assemble part", ["align", "insert", "fasten"]) as gov: while not done(): obs = sense() d = gov.step(reached=reached_from(obs), obstacles=blocking(obs)) if d.adapt: change_doer.intervene(at=d.frontier) # your logic owns the "how" else: policy.act(obs) # continue as planned
Treat frontier as where to focus, not a command. It’s the milestone label
you chose; your change-doer maps it to an actual corrective behaviour.
The server holds the live token object — its knowledge graph accumulates across
steps. Two consequences:
One session per task attempt. Open it once, step it every cycle, close it at
the end. Don’t open a fresh session per tick.
Close what you open. Sessions and anticipators persist server-side until
deleted. The with form guarantees cleanup; if you hold one by hand, close it in
a finally.
s = tk.session("goal", ["a", "b"])try: run_loop(s)finally: s.close()
megan-tk routes are rate-limited per account (default 120 requests/minute).
This is deliberate — it keeps the decision boundary from being reconstructed by
mass probing. Design your loop to stay under it.
Exceeding the limit returns HTTP 429, surfaced as MeganTKError with a
Retry-After hint in the message. Don’t hot-loop calls: for a fast control loop,
call megan-tk on a decimated cadence (e.g. every N ticks, or only when sensed
state changes) rather than every single tick.
STRIDE = 5 # govern every 5th control tickfor i, obs in enumerate(loop()): if i % STRIDE == 0: last = gov.step(reached=reached_from(obs), obstacles=blocking(obs)) apply(last) # reuse the last decision between calls
Missing/invalid/revoked key, deactivated account (401/403). Not retryable — fix the credential.
MeganTKError
Everything else: bad input (422), rate limit (429), not found (404), unreachable API.
import timefrom cadenza_cli import MeganTK, AuthRequired, MeganTKErrordef govern(session, **kw): for attempt in range(3): try: return session.step(**kw) except AuthRequired: raise # credential problem — stop except MeganTKError as e: if "rate limit" in str(e).lower() and attempt < 2: time.sleep(2 ** attempt) # back off, then retry continue raise
A 422 means your input is wrong (an out-of-range milestone index, perceive
on a non-perception session, an unregistered action name). These won’t fix
themselves on retry — correct the call.
The two primitives are independent and compose in a single loop: a session
governs task progress while an anticipator guards against a periodic
disturbance.
from cadenza_cli import MeganTKtk = MeganTK()with tk.session("carry tray", ["lift", "traverse", "place"]) as gov, \ tk.anticipator(actions=["brace", "steady"], min_events=2) as guard: for i, ev in enumerate(control_loop()): # 1) periodic-disturbance protection if ev.bump_detected: guard.disturbance(t=ev.t) p = guard.protect(t=ev.t) if p.should_protect and p.ready: saved = actuators.engage(p.best_action) guard.outcome(p.best_action, saved=saved) # 2) task governance (decimated to respect the rate limit) if i % 5 == 0: d = gov.step(reached=reached_from(ev), obstacles=blocking(ev)) if d.adapt: change_doer.intervene(at=d.frontier)
