Token sessions

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. This is how Megan does finding ways around obstacles — when the frontier is blocked, it routes you to adapt (act here) instead of letting a frozen plan push into the wall. There are two ways to drive a session:

Path	Method	You report	Open with
Milestone	`s.step(...)`	which milestones are reached / what’s blocking	milestones only
Perception	`s.perceive(...)`	robot position, goal, sensed obstacles	milestones + `ring_radii`

Opening a session

from cadenzalabs 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 cadenzalabs 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

Disturbance anticipators

Learn a periodic disturbance and protect ahead of it.

SDK reference

Every method, argument, and return type.

Introduction

Configure

Using the CLI

Megan

Reference

Opening a session

The milestone path — `step`

Acting on the decision

The perception path — `perceive`

Inspecting and closing

A full milestone loop

Next

Disturbance anticipators

SDK reference

​Opening a session

​The milestone path — step

​Acting on the decision

​The perception path — perceive

​Inspecting and closing

​A full milestone loop

​Next

Disturbance anticipators

SDK reference

Opening a session

The milestone path — `step`

Acting on the decision

The perception path — `perceive`

Inspecting and closing

A full milestone loop

Next