09_sessions.py

"""09: Sessions: capture an ID, resume it, fork it, and inspect it on disk."""

import asyncio

from claude_agent_sdk import (
    ClaudeAgentOptions,
    ResultMessage,
    get_session_info,
    get_session_messages,
    list_sessions,
    query,
    rename_session,
)
from demo_util import PROJECT


async def run(prompt, **opts):
    session_id, text = None, None
    async for message in query(prompt=prompt, options=ClaudeAgentOptions(cwd=PROJECT, max_turns=4, **opts)):
        if isinstance(message, ResultMessage):
            session_id, text = message.session_id, message.result
    return session_id, text


async def main():
    # 1. A fresh session that establishes some context
    sid, text = await run("Remember this code word: PELICAN. Reply only with 'noted'.", allowed_tools=[])
    print(f"Original session: {sid} -> {text!r}")

    # 2. Resume: the agent still knows the code word
    rid, text = await run("What was the code word? One word.", resume=sid)
    print(f"Resumed session:  {rid} -> {text!r}  (same id: {rid == sid})")

    # 3. Fork: branch into a new session that starts with the same history
    fid, text = await run("Change the code word to HERON and reply 'changed'.", resume=sid, fork_session=True)
    print(f"Forked session:   {fid} -> {text!r}  (new id: {fid != sid})")

    # 4. The original is untouched
    _, text = await run("What is the code word? One word.", resume=sid)
    print(f"Original again:        -> {text!r}")

    # 5. Inspect sessions on disk
    rename_session(sid, "codeword-demo", directory=PROJECT)
    info = get_session_info(sid, directory=PROJECT)
    print(f"\nget_session_info: title={info.custom_title!r} first_prompt={info.first_prompt!r} cwd={info.cwd}")
    recent = list_sessions(directory=PROJECT, limit=3)
    print("list_sessions (latest 3):")
    for s in recent:
        print(f"  {s.session_id}  {s.custom_title or s.summary!r}")
    msgs = get_session_messages(sid, directory=PROJECT)
    print(f"get_session_messages: {len(msgs)} messages, roles={[m.type for m in msgs]}")


asyncio.run(main())