tada-core 0.8.5

Agent execution loop SDK: streaming events, tool dispatch, compaction, and session state for Anthropic-compatible model endpoints.

tada-core

tada-core 是 tada-agent 的多协议 agent 执行内核，负责驱动模型 ↔ 工具 ↔ hook 的运行循环，并向上层暴露统一的事件流与会话状态。

核心入口：AgentLoop。模型接入推荐通过 ModelSpec + build_model_client() 选择内置 adapter。

作为 SDK 引入

阶段 A（tada-agent 本地联调）推荐使用 sibling editable path：

[project]
dependencies = ["tada-core"]

[tool.uv.sources]
tada-core = { path = "../tada-core", editable = true }

团队内 CI 或跨仓验证可锁定 Git tag：

[project]
dependencies = ["tada-core"]

[tool.uv.sources]
tada-core = { git = "ssh://git@.../tada-core.git", tag = "v0.1.0" }

阶段 B 发布到 PyPI 后再切换为：

dependencies = ["tada-core>=0.1,<0.2"]

不要把本地 path 依赖写进生产环境配置。更完整的安装说明见 docs/integration/install.md。

负责什么

• AgentLoop.run_turn() / run_turn_message() / continue_turn()：单轮 agent 执行的 while 循环
• Event：运行过程的统一事件流（text/thinking delta、tool_use、permission、hook、compaction、usage、error 等）
• SessionState：跨轮可序列化的会话状态（messages、usage、metadata）
• ConversationMessage / ContentBlock：归一化 transcript（text、thinking、tool、image、file）
• ModelSpec + adapter registry：轻量模型能力描述与多协议 adapter 选择
• tools / permissions / hooks / compaction / budget：runtime 装配点
• 内置 adapter：anthropic-messages、openai-responses、openai-completions、google-generative-ai、bedrock-converse-stream

不负责的事情（由上层实现）：

• HTTP / WebSocket / CLI / UI
• 业务会话存储与鉴权
• human-in-the-loop 交互
• hint / 系统提示拼接 / 业务级上下文选择策略
• tool middleware（参数修复、错误捕获、流式 chunk 处理）
• 多 provider 路由、fallback、账号额度、成本估算/价格表（使用 token-router 或宿主）
• OAuth 登录/刷新
• 长期记忆 / 用户档案 / 跨会话记忆检索
• 具体文件上传、下载、权限和 channel 渲染
• multi-agent 编排
• 把 Event 翻译成产品协议

安装

uv sync

最小使用

import asyncio
from tada_core import AgentLoop, SessionState
from tada_core.loop.model_client import OpenAICompatibleModelClient, make_echo_transport

loop = AgentLoop(
    OpenAICompatibleModelClient(make_echo_transport(), streaming=True),
    system_prompt=["You are a helpful assistant."],
)
state = SessionState()

async def main() -> None:
    async for event in loop.run_turn(state, "hello"):
        if event.type == "text_delta":
            print(event.text, end="", flush=True)
    print()

asyncio.run(main())

接真实模型（Anthropic via ModelSpec）：

from tada_core import AgentLoop, ModelSpec, SessionState, build_model_client

spec = ModelSpec(
    id="claude-sonnet-4-6",
    provider="anthropic",
    api="anthropic-messages",
    base_url="http://localhost:8787",
)
loop = AgentLoop(
    build_model_client(spec, api_key="token-router-key", streaming=True),
    system_prompt=["You are a helpful assistant."],
)

或直接构造（v0.1 兼容）：

from tada_core import AgentLoop, AnthropicModelClient, SessionState

loop = AgentLoop(
    AnthropicModelClient(api_key="sk-ant-...", model="claude-sonnet-4-6"),
    system_prompt=["You are a helpful assistant."],
)

核心能力

Event 事件流

AgentLoop.run_turn() 是一个异步生成器，按固定顺序 yield Event。所有事件类型通过 type 字段区分：

type	说明
turn_start	一轮 turn 开始
text_start / text_delta / text_end	助手文本块生命周期
thinking_start / thinking_delta / thinking_end	reasoning/thinking 块生命周期
tool_input_delta	工具参数 JSON 的增量片段
tool_use	模型请求调用某工具（含 tool_use_id、tool_name、input_json）
tool_result	工具执行完成（含 output、is_error 标志）
tool_progress	长耗时工具通过 on_update 报告的中间进度
usage	token 用量（input_tokens、output_tokens、cache_read_tokens、cache_creation_tokens）
message_stop	一次助手消息的逻辑结束（含 stop_reason）
permission	工具执行前的权限决策（allowed: bool、可选 reason）
hook	hook 参与循环时的观测事件（hook_name、phase、messages）
compaction	会话压缩摘要（strategy、removed_message_count、summary）
error	运行时错误（message、code）
unsupported	某能力当前不支持（feature、detail）；当前 AgentLoop 主路径暂不会主动发射
loop_exit	POST_TOOL_USE hook 请求当前 turn 优雅退出
steering	POST_TOOL_USE hook 注入 steering messages 并跳过剩余工具
turn_end	一轮 run_turn() 结束

事件顺序保证：tool_use → permission → tool_result；tool_progress 出现在对应最终 tool_result 之前。

SessionState

跨轮保存全部上下文，可序列化（Pydantic）。

state = SessionState()
# 字段：
state.messages   # list[ConversationMessage] — 完整 transcript
state.usage      # TokenUsage(input_tokens, output_tokens, cache_read_tokens, cache_creation_tokens)
state.metadata   # dict[str, Any] — 供上层写入任意 KV

多轮对话只需将同一个 state 对象传给每次 run_turn()。

Multimodal turns 与 continue

from tada_core import ContentBlock, ConversationMessage, MessageRole

msg = ConversationMessage(
    role=MessageRole.USER,
    content=[
        ContentBlock(type="text", text="describe this image"),
        ContentBlock(type="image", source="base64", mime_type="image/png", data="..."),
    ],
)
async for event in loop.run_turn_message(state, msg):
    ...

# 从已有 tool_result 继续，不追加新的 user message
async for event in loop.continue_turn(state):
    ...

# LLM 调用前裁剪/注入 context
loop.set_context_transform(lambda messages: messages[-40:])

Tools

工具通过 ToolExecutor 接口挂载，与模型协议解耦：

from tada_core.contracts.tools import InMemoryToolExecutor, ToolDefinition

ex = InMemoryToolExecutor()
ex.register(
    ToolDefinition(
        name="echo",
        description="Return input unchanged.",
        parameters_schema={"type": "object"},
    ),
    lambda inp: inp,   # 接收 JSON 字符串，返回 JSON 字符串
)
loop.set_tool_executor(ex)

• ToolDefinition：工具名、描述、JSON Schema 参数定义
• ToolExecutor.execute(tool_name, input_json) -> str：输入/输出均为 JSON 字符串
• 未知工具或执行失败应抛 ToolError，AgentLoop 会将其转为 ToolResultEvent(is_error=True)

Permissions

工具执行前调用 PermissionPolicy.authorize()，结果影响是否继续执行并产生 PermissionEvent：

from tada_core.contracts.permissions import DenyListPolicy

loop.set_permission_policy(DenyListPolicy(denied={"shell", "rm"}))

内置策略：

类	行为
AllowAllPolicy	默认，放行所有工具
DenyListPolicy	拒绝指定工具名集合（大小写不敏感）

可实现 PermissionPolicy ABC 自定义策略。PermissionContext 可将 hook 结果注入权限判断。

Runtime Budget

BudgetLimits 提供 provider-agnostic 的运行时预算控制：token、迭代次数、工具调用次数。它不包含价格表、provider 计费规则或账号额度。

from tada_core import BudgetLimits

loop.set_budget_policy(BudgetLimits(max_total_tokens=100_000, max_tool_calls=20))

如果宿主需要成本上限，应在 token-router 或宿主层把 UsageEvent 转成成本，再决定是否继续调用。

Hooks

Hook 在工具执行前后介入，可观测、拒绝、替换工具输入：

from tada_core.contracts.hooks import FunctionHookRunner, HookResult

def pre(tool_name: str, input_json: str) -> HookResult:
    if tool_name == "dangerous":
        return HookResult(denied=True, messages=["blocked"])
    return HookResult.allow()

loop.set_hook_runner(FunctionHookRunner(pre=pre))

当前 AgentLoop 会调用这些 hook 阶段：

阶段	触发时机
ON_TURN_START	run_turn() 入口
PRE_LLM_CALL	每次模型调用前，可替换 message 列表
POST_LLM_CALL	每次模型响应后，可观测文本与工具名
PRE_TOOL_USE	工具执行前
POST_TOOL_USE	工具成功执行后
POST_TOOL_USE_FAILURE	工具执行失败后
ON_TURN_END	run_turn() 退出前

HookRunner 也定义了 ON_SESSION_START，但当前 AgentLoop 还没有调用这个阶段。

HookResult 关键字段：

• denied / failed / cancelled：停止当前执行
• updated_input：替换传给工具的 input（优先于原始输入）
• updated_messages：在 PRE_LLM_CALL 阶段替换传给模型的消息列表
• exit_loop：在工具完成后请求结束当前 turn
• steering_messages：注入新的消息并让模型重新规划
• messages：出现在 HookEvent.messages 中，供上层展示

Compaction

会话超长时默认启用结构化摘要压缩。AgentLoop 会把 provider 返回的单次完整输入归一化为 TokenUsage.context_input_tokens。post_llm 优先使用该真值判断是否需要压缩； pre_turn、post_turn 和没有有效 usage 的场景估算当前请求，避免复用已经过期的 usage。

from tada_core import CompactionConfig, StructuredSummaryCompaction, TruncateCompaction

# 默认：StructuredSummaryCompaction(model)
loop = AgentLoop(model)

# 显式关闭 core 默认压缩
loop = AgentLoop(model, compaction="disabled")

# hook 接管：core 不主动压缩，但仍允许 hook 改写 messages
loop = AgentLoop(model, compaction=CompactionConfig(mode="hook_only"))

# 自定义策略
loop.set_compaction_strategy(TruncateCompaction(max_messages=20))

StructuredSummaryCompaction 的阈值默认从 ModelSpec.context_window * 0.7 派生； 宿主显式传入 threshold_tokens 时优先使用显式值。摘要请求会复用主对话 system_prompt 和 tools，让 Anthropic 显式 cache、OpenAI 隐式 prefix cache 等 provider cache 机制自然命中。

内置策略：

类	行为
NeverCompact	不压缩
TruncateCompaction	丢弃最旧消息，保留最近 N 条
SummarizeCompaction	简单截断 + 占位摘要（适合测试）
ModelSummaryCompaction	低级 LLM 摘要策略，保留兼容
StructuredSummaryCompaction	默认高质量结构化摘要，保留 head/tail、维护 tool pair、写入 manifest

Hook 接管压缩时可返回：

HookResult(
    updated_messages=compacted_messages,
    compaction_handled=True,
    skip_builtin_compaction=True,
)

这样可以避免 tada-core 默认策略与宿主（例如 tada-agent 现有 MemoryCompactionHook）发生双重压缩。更完整的语义、成本和 provider cache 行为见 docs/reference/compaction.md。

ModelClient

两种内置客户端：

AnthropicModelClient（推荐，tada-agent 默认）：

from tada_core import AnthropicModelClient

client = AnthropicModelClient(
    api_key="sk-ant-...",
    model="claude-sonnet-4-6",
    base_url="",          # 默认 https://api.anthropic.com；可设为 OpenRouter、token-router 等
    streaming=True,
    max_tokens=8192,
    extra_headers={},     # 例如 OpenRouter 的 HTTP-Referer
    cache_retention="short",  # "none" | "short" | "long"
)

• 不依赖 anthropic SDK，直接用 httpx
• 兼容任何实现 Anthropic Messages API 格式 (/v1/messages) 的端点
• cache_retention="long" 仅在 base_url 命中 api.anthropic.com 时发送 1h TTL；token-router / LiteLLM 等代理默认使用 short retention，且只打 content block 级 cache_control

OpenAICompatibleModelClient（echo 测试 / 非 Anthropic 端点）：

from tada_core.loop.model_client import OpenAICompatibleModelClient, make_echo_transport

# echo 模式（无需 API key，用于测试 loop 本身）
client = OpenAICompatibleModelClient(make_echo_transport(), streaming=True)

自定义 transport 可对接任何 OpenAI chat/completions 格式端点。

验证层级

tada-core 的功能验证分三层：

Layer 1  单元/契约测试（tests/conformance/）
         stub model，纯逻辑，< 0.1s，CI 强制通过
         uv run pytest tests/

Layer 2  真实 LLM 手动烟雾测试（scripts/smoke.py）
         AgentLoop + 真实 endpoint，手动跑，需要 API key
         验证 text_delta / tool_use / permission / usage 等完整事件链路

Layer 3  交互式 UI 验证（agent-apps/agent-studio/）
         FastAPI backend + Next.js 前端，在浏览器里看多 Agent 事件流、产物和 SessionState

Layer 1：单元测试

uv run pytest tests/
uv run ruff check tada_core tests scripts

测试覆盖：

• tests/conformance/test_agent_loop.py：AgentLoop 完整能力（工具、权限、hook、compaction、abort）
• tests/conformance/test_anthropic_client.py：Anthropic 消息序列化与流式解析（offline）
• tests/conformance/test_contract_semantics.py：SessionState、事件顺序、permission context
• tests/conformance/test_event_sequences.py：事件序列顺序保证
• tests/conformance/test_clawcode_model_client.py：OpenAI-compatible 客户端解析

Layer 2：烟雾测试

scripts/smoke.py 是一个独立脚本，直接驱动 AgentLoop，把每个 Event 打印出来。

# echo 模式（无需 API key，验证 loop 本身）
uv run python scripts/smoke.py

# 接真实模型（Anthropic direct 或任何兼容代理）
API_KEY=<key> \
BASE_URL=https://your-proxy-host \
MODEL_ID=claude-sonnet-4-6 \
uv run python scripts/smoke.py

# 多轮对话（每个 --text 是一轮，共享同一个 SessionState）
API_KEY=<key> MODEL_ID=claude-haiku-4-5-20251001 \
uv run python scripts/smoke.py --text "what is 2+2" --text "multiply that by 3"

环境变量说明：

变量	说明
API_KEY	API key，不填则 echo 模式
MODEL_ID	模型 ID，默认 claude-sonnet-4-6
BASE_URL	接入端点，默认 https://api.anthropic.com
PROTOCOL	anthropic（默认）或 openai

Layer 3：交互式 UI 验证

bash agent-apps/agent-studio/dev.sh

打开 http://127.0.0.1:3010，在 Agent Studio 里选择 agent、发送消息，并在 Inspector 中观察 trace、artifact 和运行状态。

不配置模型则使用 echo 模式，在"设置"页填入 API key 后走真实 LLM。

目录结构

tada_core/
  contracts/        纯数据结构与接口（无 IO、无网络）
    events.py         Event union（15 种事件类型）
    session.py        SessionState、TokenUsage
    messages.py       ConversationMessage、ContentBlock、MessageRole
    model.py          ModelClient ABC、ApiRequest、AssistantEvent
    tools.py          ToolDefinition、ToolExecutor ABC、InMemoryToolExecutor
    permissions.py    PermissionPolicy ABC、AllowAllPolicy、DenyListPolicy
    hooks.py          HookRunner ABC、FunctionHookRunner、HookResult
    compaction.py     CompactionStrategy ABC、4 种内置策略
  loop/             AgentLoop 执行循环 + model clients
    runner.py         AgentLoop（唯一的 while 循环实现）
    model_client.py   AnthropicModelClient、OpenAICompatibleModelClient
  extensions/       可选扩展（MCP）
  cli/              echo-only 本地调试入口（tada-core-smoke）
scripts/
  smoke.py          Layer 2 烟雾测试脚本
tests/
  conformance/      Layer 1 契约测试
agent-apps/
  agent-studio/     多 Agent 统一宿主（推荐）
    backend/          FastAPI + 插件 registry
    frontend/         三栏 UI：Chat / Trace / Artifact
  shared/
    sandbox/          coding-agent 复用的 sandbox 包、fixtures、runtime 与回归测试
  agent-studio/dev.sh  推荐启动入口
docs/               架构与接入文档
AGENTS.md           AI / coding agent 机器可读入口

文档

• docs/README.md：文档地图
• AGENTS.md：AI agent 接入入口