会话管理¶

import asyncio
from claude_agent_sdk import (
    ClaudeSDKClient,
    ClaudeAgentOptions,
    AssistantMessage,
    ResultMessage,
    TextBlock,
)


def print_response(message):
    """只打印人类可读的消息部分"""
    if isinstance(message, AssistantMessage):
        for block in message.content:
            if isinstance(block, TextBlock):
                print(block.text)
    elif isinstance(message, ResultMessage):
        cost = (
            f"${message.total_cost_usd:.4f}"
            if message.total_cost_usd is not None
            else "N/A"
        )
        print(f"[完成: {message.subtype}, 花费: {cost}]")


async def main():
    options = ClaudeAgentOptions(
        allowed_tools=["Read", "Edit", "Glob", "Grep"],
    )

    async with ClaudeSDKClient(options=options) as client:
        # 第一次查询：Client 内部自动捕获 session ID
        await client.query("分析认证模块")
        async for message in client.receive_response():
            print_response(message)

        # 第二次查询：自动继续同一个会话，无需传 ID
        await client.query("现在把它重构为使用 JWT")
        async for message in client.receive_response():
            print_response(message)


asyncio.run(main())

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

async def main():
    session_id = None

    async for message in query(
        prompt="分析认证模块并建议改进",
        options=ClaudeAgentOptions(
            allowed_tools=["Read", "Glob", "Grep"],
        ),
    ):
        if isinstance(message, ResultMessage):
            session_id = message.session_id
            if message.subtype == "success":
                print(message.result)

    print(f"Session ID: {session_id}")
    return session_id


session_id = asyncio.run(main())

# 之前的会话已经做了分析，现在直接基于分析结果行动
async for message in query(
    prompt="现在实现你建议的重构方案",
    options=ClaudeAgentOptions(
        resume=session_id,
        allowed_tools=["Read", "Edit", "Write", "Glob", "Grep"],
    ),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

# Fork：从 session_id 分叉出新会话
forked_id = None
async for message in query(
    prompt="不用 JWT，改用 OAuth2 实现认证模块",
    options=ClaudeAgentOptions(
        resume=session_id,
        fork_session=True,
    ),
):
    if isinstance(message, ResultMessage):
        forked_id = message.session_id  # Fork 的新 ID，与 session_id 不同
        if message.subtype == "success":
            print(message.result)

print(f"分叉会话: {forked_id}")

# 原始会话完好无损，恢复它继续 JWT 方案
async for message in query(
    prompt="继续 JWT 方案",
    options=ClaudeAgentOptions(resume=session_id),
):
    if isinstance(message, ResultMessage) and message.subtype == "success":
        print(message.result)

from claude_agent_sdk import ClaudeSDKClient, ClaudeAgentOptions

options = ClaudeAgentOptions(
    enable_file_checkpointing=True,
    permission_mode="acceptEdits",
    extra_args={"replay-user-messages": None},
)

async with ClaudeSDKClient(options) as client:
    await client.query("重构认证模块")

checkpoint_id = None
session_id = None

async for message in client.receive_response():
    # 每条用户消息更新检查点（只保留最新的）
    if isinstance(message, UserMessage) and message.uuid:
        checkpoint_id = message.uuid
    # 从 ResultMessage 捕获 session ID
    if isinstance(message, ResultMessage):
        session_id = message.session_id

async with ClaudeSDKClient(
    ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
) as client:
    await client.query("")  # 空 prompt 建立连接
    async for message in client.receive_response():
        await client.rewind_files(checkpoint_id)
        break

safe_checkpoint = None

async with ClaudeSDKClient(options) as client:
    await client.query("重构认证模块")

    async for message in client.receive_response():
        # 每次回合前更新检查点（只保留最新的）
        if isinstance(message, UserMessage) and message.uuid:
            safe_checkpoint = message.uuid

        # 根据你的逻辑决定是否回退
        # 例如：错误检测、验证失败、用户输入等
        if your_revert_condition and safe_checkpoint:
            await client.rewind_files(safe_checkpoint)
            break

from dataclasses import dataclass
from datetime import datetime


@dataclass
class Checkpoint:
    id: str
    description: str
    timestamp: datetime


checkpoints = []
session_id = None

async with ClaudeSDKClient(options) as client:
    await client.query("重构认证模块")

    async for message in client.receive_response():
        if isinstance(message, UserMessage) and message.uuid:
            checkpoints.append(
                Checkpoint(
                    id=message.uuid,
                    description=f"第 {len(checkpoints) + 1} 轮之后",
                    timestamp=datetime.now(),
                )
            )
        if isinstance(message, ResultMessage) and not session_id:
            session_id = message.session_id

# 之后：恢复到任意检查点
if checkpoints and session_id:
    target = checkpoints[0]  # 选择任意检查点
    async with ClaudeSDKClient(
        ClaudeAgentOptions(enable_file_checkpointing=True, resume=session_id)
    ) as client:
        await client.query("")
        async for message in client.receive_response():
            await client.rewind_files(target.id)
            break
    print(f"已回退到: {target.description}")