SDK:会话浏览器

构建可浏览 Agent 会话历史的工具

🌐 查看英文原文 | 源码 Notebook


构建会话浏览器

当你将 Agent 作为产品交付时(桌面应用、IDE 扩展、内部聊天机器人),用户首先要求的就是侧边栏。他们想看到上周二的对话,重新回到其中,并且可能在不丢失原始对话的情况下分支到新的方向。Agent 循环是产品的一半;另一半是会话管理。

Claude Agent SDK 将每个对话写入磁盘上的 JSONL 记录。它还提供了一组函数来读取这些记录并组织它们,这样你就可以构建那个侧边栏,而无需编写文件解析器或手动遍历 ~/.claude/projects/。

通过本 cookbook,你将能够:

  • 列出和渲染项目的过去会话,带有分页和元数据如分支、标题和最后修改时间
  • 将存储会话的消息读回你的 UI,而无需启动 Agent
  • 重命名、标记和过滤会话,以便用户可以组织他们的历史记录
  • 在任何时刻分叉会话,并将分叉恢复为实时 query() 调用

这是 Claude Code Desktop 和 VS Code 扩展中会话侧边栏背后的模式。相同的原语适用于你想在 Agent SDK 之上构建的任何 UI。

前提条件

在遵循本指南之前,请确保你具备:

所需知识

  • Python 基础知识,包括 async/await
  • 对 Agent SDK 的 query() 函数有基本了解(参见 Notebook 00 的介绍)

所需工具

  • Python 3.11 或更高版本
  • Claude Code CLI(npm install -g @anthropic-ai/claude-code)
  • Anthropic API 密钥(在此获取)

环境配置

安装所需的依赖。会话管理函数在 claude-agent-sdk v0.1.51 中引入。

%%capture
%pip install -U "claude-agent-sdk>=0.1.51" python-dotenv pandas

从 .env 加载你的 API 密钥并配置模型。我们在这里使用 Haiku,因为演示会话很短,我们希望它们便宜且快速;在真正的产品中,你会选择适合你 Agent 的任何模型。

import os
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

MODEL = "claude-haiku-4-5"

# All demo sessions live under this project directory. Using a dedicated
# cwd keeps the demo isolated from your real Claude Code sessions.
# Note: this path resolves relative to the kernel's working directory
# (claude_agent_sdk/ when launched per the README).
DEMO_DIR = str(Path("session_browser_demo").resolve())
os.makedirs(DEMO_DIR, exist_ok=True)
print(f"Demo project dir: {DEMO_DIR}")

第一部分:创建一些要管理的会话

会话管理函数读取 query()(或 Claude Code CLI)已经写入的记录。为了有东西可以浏览,我们首先运行三个简短的对话并捕获它们的 session ID。

cwd 选项告诉 SDK 此对话属于哪个项目目录。记录归档在 ~/.claude/projects/<encoded-cwd>/<session-id>.jsonl 下,因此每次使用相同 cwd 的调用最终都在同一个桶中。那个桶就是 list_sessions(directory=...) 稍后读取的内容。

我们禁用工具并将每次运行限制为一轮,以最小化 token 使用。

from claude_agent_sdk import ClaudeAgentOptions, ResultMessage, query


async def run_one_turn(prompt: str) -> str:
    """Run a single-turn conversation and return its session_id."""
    opts = ClaudeAgentOptions(
        model=MODEL,
        cwd=DEMO_DIR,
        max_turns=1,
        allowed_tools=[],  # text-only, no tool loop
    )
    session_id = None
    async for msg in query(prompt=prompt, options=opts):
        if isinstance(msg, ResultMessage):
            session_id = msg.session_id
            preview = (msg.result or "")[:80]
            print(f"[{session_id[:8]}] {preview}...")
    if session_id is None:
        raise RuntimeError("No ResultMessage received; check API key and SDK version.")
    return session_id

三个 prompt,三个会话。在真正的产品中,这些将是你的用户向 Agent 提出的任何问题。

prompts = [
    "Give me three name ideas for a CLI tool that manages git worktrees.",
    "Explain the difference between a mutex and a semaphore in one paragraph.",
    "Write a haiku about merge conflicts.",
]

demo_session_ids = []
for p in prompts:
    sid = await run_one_turn(p)
    demo_session_ids.append(sid)

print(f"\nCreated {len(demo_session_ids)} sessions.")

第二部分:列出和检查会话

构建会话列表

list_sessions() 扫描项目的记录目录并返回每个会话的元数据,按最新排序。它读取文件统计信息以及每个记录的开头和结尾,因此即使目录中有数百个文件,它也能保持快速。不会生成子进程,也不会进行 API 调用。

每个 SDKSessionInfo 携带在选择器中渲染一行所需的信息:显示摘要、最后修改时间戳、git 分支、工作目录,以及你设置的任何自定义标题或标签。

from datetime import datetime

import pandas as pd

from claude_agent_sdk import SDKSessionInfo, list_sessions

sessions = list_sessions(directory=DEMO_DIR)

# Render as a table. In a real UI this would be your sidebar component.
rows = []
for s in sessions:
    rows.append(
        {
            "id": s.session_id[:8],
            "summary": (s.summary[:50] + "...") if len(s.summary) > 50 else s.summary,
            "modified": datetime.fromtimestamp(s.last_modified / 1000).strftime("%H:%M:%S"),
            "branch": s.git_branch or "-",
            "tag": s.tag or "-",
        }
    )

pd.DataFrame(rows)

对于较长的历史记录,传递 limit 和 offset 来分页浏览结果。会话选择器通常加载第一页,然后在用户滚动时获取更多:

page_2 = list_sessions(directory=DEMO_DIR, limit=20, offset=20)

如果你的应用已经存储了 session ID(例如,在你自己的数据库中与用户记录一起),而你只需要那一行,get_session_info() 比列出所有内容更便宜。

from claude_agent_sdk import get_session_info

info = get_session_info(demo_session_ids[0], directory=DEMO_DIR)

print(f"Session:      {info.session_id}")
print(f"Summary:      {info.summary}")
print(f"First prompt: {info.first_prompt}")
print(f"Created:      {datetime.fromtimestamp(info.created_at / 1000)}")
print(f"Size:         {info.file_size:,} bytes")

读取会话的消息

当用户在侧边栏中点击会话时,你将其对话加载到主视图中。get_session_messages() 从记录中重建消息链,并按顺序返回用户和助手的轮次。与列表函数一样,它是纯文件读取:Agent 不需要运行。

每个 SessionMessage 有一个 type("user" 或 "assistant")、一个 uuid,以及一个与 Anthropic Messages API 相同形状的 message 字典(role、content)。

from claude_agent_sdk import get_session_messages

messages = get_session_messages(demo_session_ids[0], directory=DEMO_DIR)

for m in messages:
    role = m.type
    # content is a list of blocks; pull out the text ones
    text_parts = [
        b.get("text", "")
        for b in m.message.get("content", [])
        if isinstance(b, dict) and b.get("type") == "text"
    ]
    text = " ".join(text_parts).strip()
    print(f"[{role:>9}] {text[:100]}")

对于较长的会话,limit 和 offset 让你每次加载一个窗口。聊天视图可能在打开时加载最后 50 条消息,然后在用户向上滚动时获取更旧的页面。偏移量按时间顺序应用(最旧的在前),所以第 0 页是对话的开始。

第三部分:用标题和标签组织

重命名会话

自动生成的摘要适合快速浏览,但用户通常想给会话起一个真正的名字。rename_session() 向记录追加一个标题条目;list_sessions() 在下次读取时将其作为 custom_title 获取。

追加是廉价且幂等的:调用 rename 两次只意味着较新的标题获胜。不会发生文件重写。

from claude_agent_sdk import rename_session

rename_session(demo_session_ids[0], "Worktree CLI naming brainstorm", directory=DEMO_DIR)
rename_session(demo_session_ids[2], "Haiku corner", directory=DEMO_DIR)

# Verify the titles stuck
for s in list_sessions(directory=DEMO_DIR):
    label = s.custom_title or "(auto)"
    print(f"{s.session_id[:8]}  custom_title={label!r}  summary={s.summary[:40]!r}")

标记和过滤

标签是附加到会话的单个字符串。将它们用于你的产品所需的任何分类:"archived"、"needs-review"、"favorite"。传递 None 来清除标签。

一种常见的模式是软删除:不要删除记录文件,而是将其标记为 "__hidden" 并在列表视图中过滤掉。数据保持可恢复。

from claude_agent_sdk import tag_session

# Mark two sessions as favorites, hide the other
tag_session(demo_session_ids[0], "favorite", directory=DEMO_DIR)
tag_session(demo_session_ids[2], "favorite", directory=DEMO_DIR)
tag_session(demo_session_ids[1], "__hidden", directory=DEMO_DIR)


def visible_sessions(directory: str, tag_filter: str | None = None) -> list[SDKSessionInfo]:
    """List sessions, hiding soft-deletes and optionally filtering by tag."""
    results = []
    for s in list_sessions(directory=directory):
        if s.tag == "__hidden":
            continue
        if tag_filter is not None and s.tag != tag_filter:
            continue
        results.append(s)
    return results


favorites = visible_sessions(DEMO_DIR, tag_filter="favorite")
print(f"Visible favorites: {len(favorites)}")
for s in favorites:
    print(f"  {s.session_id[:8]}  [{s.tag}]  {s.custom_title or s.summary}")

标签是单个值,不是列表。如果你需要多个维度(例如,状态加类别),将它们编码为一个字符串如 "review:urgent" 并在读取时解析,或者在以 session_id 为键的自己的数据库中存储更丰富的状态。

第四部分:分叉和恢复

从现有对话分支

分叉将会话的记录复制到一个新文件中,并重新映射消息 ID。原始会话保持不变。这是”尝试不同方法”功能背后的原语:用户保留他们的原始线程并获得一个新的线程来实验。

fork_session() 写入新文件并返回其 ID。它不运行 Agent,因此分叉会留在磁盘上,直到你用 query() 恢复它。

from claude_agent_sdk import fork_session

source = demo_session_ids[0]

fork = fork_session(
    source,
    directory=DEMO_DIR,
    title="Worktree CLI names (round 2)",
)

print(f"Source: {source}")
print(f"Fork:   {fork.session_id}")

# The fork starts with the same message history as the source
source_msgs = get_session_messages(source, directory=DEMO_DIR)
fork_msgs = get_session_messages(fork.session_id, directory=DEMO_DIR)
print(f"Source has {len(source_msgs)} messages, fork has {len(fork_msgs)}")

要从特定点而不是完整历史分支,传递 up_to_message_id。分叉将包含源记录直到并包括该消息。你可以从 get_session_messages()[i].uuid 获取消息 UUID。

将分叉恢复为实时查询

分叉只是一个记录文件。要将其变回运行中的对话,将其 ID 交给 ClaudeAgentOptions.resume。Agent 加载分叉的历史并从那里继续。

resume_opts = ClaudeAgentOptions(
    model=MODEL,
    cwd=DEMO_DIR,
    max_turns=1,
    allowed_tools=[],
    resume=fork.session_id,
)

async for msg in query(
    prompt="Those were okay. Give me three more names, but punnier.",
    options=resume_opts,
):
    if isinstance(msg, ResultMessage):
        print(f"[fork {fork.session_id[:8]} resumed]")
        print(msg.result)

原始会话仍然保持不变。再次列表,你会看到源和分叉作为具有独立历史的单独行。

for s in list_sessions(directory=DEMO_DIR):
    marker = "(fork)" if s.session_id == fork.session_id else "      "
    print(f"{marker} {s.session_id[:8]}  {s.custom_title or s.summary[:50]}")

清理

delete_session() 移除一个记录文件。这是硬删除,这就是为什么第三部分中的软删除标签模式通常是面向用户 UI 的更安全默认选择。

在这里我们使用它来清理演示创建的所有内容。

from claude_agent_sdk import delete_session

# Clean up every session in the demo dir, including the fork
for s in list_sessions(directory=DEMO_DIR):
    delete_session(s.session_id, directory=DEMO_DIR)
    print(f"Deleted {s.session_id[:8]}")

remaining = list_sessions(directory=DEMO_DIR)
print(f"\n{len(remaining)} session(s) remaining.")

回顾

我们基于 Agent SDK 的本地记录存储构建了会话浏览器的核心:

  • 使用 list_sessions() 列表提供了渲染侧边栏所需的一切,而且它可以扩展,因为它读取文件统计信息和开头/结尾切片而不是解析整个记录。
  • 使用 get_session_messages() 读取可以加载对话以供显示,而无需启动 Agent。
  • 使用 rename_session() 和 tag_session() 组织追加元数据条目,因此它是廉价的,最近的调用获胜。
  • 使用 fork_session() 加 options.resume 分叉让用户分支对话并继续,而不影响原始对话。

所有这些都是对 ~/.claude/projects/ 的纯文件操作。无论 Agent 子进程是否运行,它们都能工作,并且它们看到与 Claude Code CLI 写入的相同的记录。

接下来去哪里

  • 连接到 UI。 这些函数与 UI 无关;将它们放在 FastAPI 路由或 Electron IPC 处理程序后面,你就有了会话侧边栏的后端。
  • 跨主机会话。 记录存在本地磁盘上。要跨机器共享会话,请同步文件或将 session ID 和消息索引到你自己的存储中。有关模式,请参阅在磁盘上管理会话。
  • TypeScript。 相同的 API 存在于 @anthropic-ai/claude-agent-sdk 中,使用 camelCase 名称(listSessions、forkSession 等)。请参阅 TypeScript SDK 参考。
  • 更大的图景。 Notebook 00 到 03 涵盖了构建 Agent 本身。本 notebook 涵盖了管理它留下的内容。

Tip

中文读者提示:本章节的代码和输出为英文原文。如需理解具体实现细节,可参考上方中文导读和代码注释。如有疑问,欢迎在 GitHub Issues 讨论。