Skip to main content
A token session is megan-tk pursuing one objective across an ordered chain of milestones. You open it once, then drive it each control cycle. It hands back a Decision: continue or adapt, and the milestone its attention is on. There are two ways to drive a session:
PathMethodYou reportOpen with
Milestones.step(...)which milestones are reached / what’s blockingmilestones only
Perceptions.perceive(...)robot position, goal, sensed obstaclesmilestones + ring_radii

​
Opening a session

from cadenza_cli import MeganTK

tk = MeganTK()

s = tk.session(
    objective="pick up the red cube",
    milestones=["reach", "grasp", "lift"],
)
print(s.session_id, s.milestones, s.perception)   # perception == False
session() returns a MeganSession. It’s a context manager — prefer with so the session is always closed on the server: 
with tk.session("pick up the red cube", ["reach", "grasp", "lift"]) as s:
    ...
# session deleted automatically here

​
The milestone path — step

Each cycle, tell the token which milestone indices are now satisfied and which new obstacles block a milestone. It returns the decision. 
d = s.step(
    reached=[0, 1],                                   # milestones 0 and 1 done
    obstacles=[{"id": "wall", "milestone": 2}],       # something blocks milestone 2
)

d.route          # 'continue' or 'adapt'
d.frontier       # e.g. 'milestone_2' — where attention is
d.session_step   # monotonically increasing step counter
d.adapt          # True when route == 'adapt'
Obstacles accept either a dict or a (id, milestone_index) tuple — both normalise to the same request: 
s.step(reached=[0], obstacles=[{"id": "wall", "milestone": 1}])
s.step(reached=[0], obstacles=[("wall", 1)])          # equivalent
reached and every obstacle’s milestone are indices into the milestone list you opened the session with. An out-of-range index is rejected by the server (HTTP 422 → MeganTKError).

​
Acting on the decision

The whole point is the adapt branch — that’s megan-tk telling you the current plan won’t carry the frontier and your change-doer should intervene here: 
d = s.step(reached=reached_now, obstacles=sensed)
if d.adapt:
    change_doer.intervene(at=d.frontier)   # re-plan / switch skill / correct
else:
    policy.continue_as_planned()

​
The perception path — perceive

For grounded navigation you can let the token reason about distance instead of discrete “reached” flags. Open the session with one shrinking ring radius per milestone, then report raw position/goal/obstacles. 
s = tk.session(
    objective="navigate to the dock",
    milestones=["approach", "dock"],
    ring_radii=[1.0, 0.3],                 # one radius per milestone, shrinking
    obstacle_block_radius=0.8,             # how close an obstacle must be to block
)
assert s.perception is True

d = s.perceive(
    position=[0.0, 0.0],                   # robot world position (xy or xyz)
    goal=[2.0, 0.0],                       # goal world position
    obstacles=[{"id": "cone", "position": [1.0, 0.0]}],
)
print(d.route, d.frontier)
Perception obstacles take a dict or an (id, position) tuple: 
s.perceive(position=[1.9, 0.0], goal=[2.0, 0.0], obstacles=[("cone", [1.0, 0.0])])
ring_radii must have exactly one radius per milestone and must shrink as milestones advance (each ring tighter than the last). Calling perceive on a session opened without ring_radii is a 422 — use step there instead.

​
Inspecting and closing

s.status()   # {'session_id', 'session_step', 'objective', 'perception'}
s.close()    # delete server-side; returns True (idempotent — second call False)
The server holds the live token object (its knowledge graph accumulates across steps), so close sessions you’re done with. The with form does this for you; if you manage a session by hand, call close() in a finally.

​
A full milestone loop

from cadenza_cli import MeganTK

tk = MeganTK()
with tk.session("pick up the red cube", ["reach", "grasp", "lift"]) as s:
    reached = []
    for obs in perception_stream():            # your sensor loop
        reached = update_reached(reached, obs)
        d = s.step(reached=reached, obstacles=blocking(obs))
        if d.adapt:
            change_doer.intervene(at=d.frontier)
        if len(reached) == len(s.milestones):
            break

​
Next

Disturbance anticipators

Learn a periodic disturbance and protect ahead of it.

SDK reference

Every method, argument, and return type.