Session Discovery

From SDK 0.3.0 the SDK wraps the runtime's ListSessions and WatchSessions RPCs. Together they let orchestrators and supervisor agents enumerate active sessions and react to lifecycle events (CREATED / RESOLVED / EXPIRED, plus CANCELLED / SUSPENDED / RESUMED since SDK 0.4.0) without polling GetSession.

For the underlying RPC contracts (request/response shapes, scoping rules), see Runtime API § Discovery and § Streaming Watches.

When to use

• Supervisor dashboards — show every OPEN session for a tenant and their terminal outcomes as they occur.
• Late-joining orchestrators — recover in-flight sessions after a restart without keeping an out-of-band session registry.
• Reconciliation — cross-check your control-plane's view of sessions against what the runtime actually accepted.

list_sessions (snapshot)

from macp_sdk import AuthConfig, MacpClient

client = MacpClient(
    target="runtime:50051",
    auth=AuthConfig.for_bearer("tok-ops", expected_sender="ops"),
)
client.initialize()

for meta in client.list_sessions():
    print(
        meta.session_id,
        meta.mode,
        meta.state,
        meta.context_id or "-",
        list(meta.extension_keys),
    )

Each entry is a SessionMetadata proto with the same shape returned by GetSession — including the projected context_id and extension_keys fields that surface any extension blobs the initiator attached to SessionStart.extensions (see Protocol → SessionStart).

Pagination (macp-proto 0.1.6)

macp-proto 0.1.6 added page_size / page_token to the ListSessions request, and the SDK threads them: list_sessions() auto-paginates — it follows the runtime's next_page_token until empty and returns the complete list — so callers are forward-compatible with a paginating runtime.

Runtime status: runtime v0.5.0 does not implement pagination server-side yet — it ignores page_size/page_token and returns the full set in a single page with an empty next_page_token. list_sessions() already returns the complete list either way; multi-page behaviour becomes observable only once a runtime honours page_size.

all_sessions = client.list_sessions(page_size=100)   # drains all pages

If you want to page manually — e.g. to render one page at a time — use list_sessions_page, which returns (sessions, next_page_token). An empty token means the last page; don't assume a complete list until the token is empty. (Against runtime v0.5.0 the loop below runs zero iterations — the first call already carries everything.)

page, token = client.list_sessions_page(page_size=50)
while token:
    more, token = client.list_sessions_page(page_size=50, page_token=token)
    page.extend(more)

SessionLifecycleWatcher (live stream)

from macp_sdk import SessionLifecycleWatcher

watcher = SessionLifecycleWatcher(client)

for event in watcher.changes():
    print(event.event_type, event.session.session_id)
    if event.is_terminal:
        # This session will not emit more events.
        ...

event.event_type is a short string ("CREATED", "RESOLVED", "EXPIRED", and — since SDK 0.4.0 / macp-proto 0.1.3 — "CANCELLED", "SUSPENDED", "RESUMED"). Convenience predicates:

Cancellation moved (SDK 0.4.0): an accepted CancelSession now surfaces as CANCELLED, not EXPIRED. is_terminal includes CANCELLED, so loops that wait until event.is_terminal keep working — but switch any code that special-cased is_expired to detect cancellation over to is_cancelled. SUSPENDED / RESUMED are non-terminal.

Suspension cap (SDK 0.5.0 / runtime v0.5.0): pass max_suspend_ms to session.start(...) (or SessionStart) to bind a per-session maximum suspension window. 0 (default) selects the runtime default (currently 7 days). A suspension that outlasts the cap expires the session (SUSPENDED → EXPIRED), which you will observe as an is_expired lifecycle event. The resolved cap is recorded on the session's SessionStart log entry and used on replay, so it is stable regardless of later runtime configuration. Negative values are rejected client-side.

Startup snapshot semantics

The runtime emits an initial CREATED event for every session that is already OPEN at subscribe time, then live events thereafter. That means a freshly-started supervisor sees every in-flight session without a separate list_sessions() call:

for event in SessionLifecycleWatcher(client).changes():
    if event.is_created:
        register(event.session)   # fires once per pre-existing session, plus every new one
    elif event.is_terminal:
        finalise(event.session)

list_sessions() is still useful when you want a bounded snapshot without holding the stream open.

Blocking handler form

watch(handler) is shorthand for a blocking for-loop:

def on_event(ev):
    dashboard.update(ev.session.session_id, ev.event_type)

SessionLifecycleWatcher(client).watch(on_event)  # blocks

Threading

The watcher reads from a gRPC server-streaming RPC on the caller's thread. Run it in a background thread or process if your agent also needs to send envelopes on the same client:

import threading

def run_watcher():
    for ev in SessionLifecycleWatcher(client).changes():
        handle(ev)

threading.Thread(target=run_watcher, daemon=True).start()

Authorisation

Both RPCs require the same Bearer auth as any other SDK call — the runtime scopes results to the authenticated identity. An agent only sees sessions it is a participant in, plus any sessions its token is authorised to observe via runtime config.

Related

• Streaming → Watchers for the full watcher catalogue (PolicyWatcher, SignalWatcher, …).
• Building Orchestrators → Supervisor pattern for a worked example.