agenthold

agenthold

阻止 AI Agent 静默覆盖彼此的数据。

当两个 Agent 更新同一个值时，后一次写入会悄无声息地覆盖前一次写入。没有错误，没有异常，只有错误的数据，以及仍在运行的系统。agenthold 是一个 MCP 服务，为 Agent 提供带有内置冲突检测的共享版本化状态。可以把它看作 Agent 工作记忆的 git。

---

问题

当两个 Agent 同时更新同一个值时，后一次写入会悄无声息地覆盖前一次写入。不会抛出异常。值是错的，但系统仍在运行。

两个 Agent 都读取了 $10,000 的预算，并分别据此进行分配。承诺总额：$15,000。预算对象毫无提示。这就是读-修改-写冲突：每个 Agent 都假设自读取后没有任何变化，然后进行写入。

---

工作原理

agenthold 通过**乐观并发控制（OCC）**解决这个问题，Postgres 在 UPDATE ... WHERE version = N 中以及 DynamoDB 在条件写入中使用的也是这种机制。

agenthold 中存储的每个值都有一个版本号。Agent 写入时，会传入读取该值时的版本。如果读取后存储的版本已发生变化，写入就会被拒绝，并返回包含当前值的 ConflictError。Agent 重新读取、重新计算，然后重试。

未能成功写入的 Agent 会检测到冲突，重新读取实际剩余预算（$2,000），并调整分配。承诺总额始终恰好为 $10,000。每次写入都会被记录。

OCC 非常适合 Agent 工作流，原因如下：

• Agent 会在读取和写入之间执行操作（网络调用、大模型推理）。你无法在这些操作期间一直持有数据库锁。
• 冲突很少见。重试一次比每次读取都获取锁成本更低。
• 重试逻辑简单、明确，完全由 Agent 控制。

---

适用于任何 Agent 框架

agenthold 通过 MCP (Model Context Protocol) 连接，这是一种用于工具集成的开放标准。任何支持 MCP 的框架都可以直接使用 agenthold，无需编写胶水代码。

框架	连接方式
Claude Desktop / Claude Code	内置：添加到 mcpServers 配置中
Cursor / Continue / Windsurf	内置：添加到 MCP 配置中
LangChain / LangGraph	langchain-mcp-adapters
CrewAI	Agent 上的原生 mcps 字段
OpenAI Agents SDK	内置 mcp_servers 参数
Google ADK	内置 MCP Toolbox
AutoGen	autogen_ext.tools.mcp
PydanticAI	原生 MCP 集成

agenthold 不是框架，而是位于编排层下方的共享基础设施，就像数据库位于应用程序下方一样。你的 Agent 保留现有工具和逻辑；agenthold 则为它们补上缺少的协调机制。

尚未使用 MCP？ agenthold 也可作为 Python 库，供任何框架直接调用。导入 StateStore，调用 .get() 和 .set() 并进行版本检查，即可获得可防止冲突的共享状态。

---

架构

graph LR
    A1["Agent 1
    LangChain, CrewAI, etc."] -->|MCP| S["agenthold
    MCP Server"]
    A2["Agent 2
    Claude, OpenAI, etc."] -->|MCP| S
    A3["Agent 3
    AutoGen, ADK, etc."] -->|MCP| S
    S --> DB[("SQLite
    WAL mode")]
    DB -->|version 3| S
    S -->|"conflict! retry"| A2

每次写入都带有版本号。如果自 Agent 读取数据后，存储的版本发生了变化，写入就会被拒绝，Agent 会使用当前数据重试。这与 Postgres 条件更新和 DynamoDB 条件写入使用的机制相同。

---

快速开始

1. 安装

pip install agenthold
# or
uv pip install agenthold

2. 添加到 MCP 客户端配置

{
  "mcpServers": {
    "agenthold": {
      "command": "agenthold",
      "args": ["--db", "/path/to/state.db"]
    }
  }
}

3. 完成

Agent 会自动协调。无需 CLAUDE.md，无需修改系统提示词，也无需设计命名空间。

Agent 连接后，会看到五个可自我说明的工具：agenthold_register、agenthold_claim、agenthold_release、agenthold_status 和 agenthold_wait。工具描述会告诉 Agent 每个工具的使用时机和方法。如果 MCP 客户端包含服务器说明，服务器说明也会进一步强化此协议。

---

资源

agenthold 使用规范 URI 标识资源。工具接受以下任一形式：

• 裸路径（例如 "src/main.py"）——解析到名为 default 的工作区；如果恰好只配置了一个工作区，则解析到该工作区。
• 显式 URI——文件使用 "file://<workspace>/<path>"，不透明资源使用 "custom://<name>"。

等效输入（./src/main.py、src\main.py、src//main.py、工作区内的绝对路径）都会规范化为相同的内部 URI，因此两个 Agent 使用不同的简写形式也不会导致键空间分裂。路径遍历（..）和点段（.）会在边界处被拒绝。

在默认情况下不区分大小写的文件系统（Windows 和 macOS）上，文件资源匹配时不区分大小写，因此 src/Main.py 和 src/main.py 会解析为同一资源——两个 Agent 无法同时认领同一个物理文件。这遵循运行平台的规则；custom:// 名称始终区分大小写。

使用 --workspace name=path 配置工作区（可重复指定）。如果未指定此标志，agenthold 会在当前工作目录中创建一个名为 default 的工作区。多工作区配置可让一个 agenthold 进程协调多个独立代码库。

---

工具

agenthold 默认提供五个协调工具。

agenthold_register

注册自己并获取唯一的 Agent ID。使用 agenthold_claim 或 agenthold_release 前必须调用一次。

{ "name": "editor-agent", "model": "claude-sonnet-4-6" }

{
  "status": "registered",
  "agent_id": "agent-a1b2c3d4",
  "name": "editor-agent",
  "registered_at": "2026-03-18T10:00:00+00:00"
}

---

agenthold_claim

在修改资源前，先声明对它的独占访问权。需要已注册的 agent_id。

{ "resource": "intro.md", "agent_id": "agent-a1b2c3d4" }

已声明（你拥有独占访问权）：

{ "status": "claimed", "resource": "file://default/intro.md", "version": 1 }

如果资源有过非简单的历史记录（已删除、已移动、已放弃或已过期），响应还会包含 previous_outcome、previous_holder、previous_outcome_at 和一个描述前一个持有者所做操作的 hint。当 previous_outcome: "moved" 时，moved_to 是新的资源 URI。

繁忙（另一个 Agent 正在处理此资源）：

{
  "status": "busy",
  "resource": "file://default/intro.md",
  "held_by": "agent-e5f6g7h8",
  "claimed_at": "2026-03-18T10:00:00+00:00",
  "hint": "Another agent holds this resource. Work on a different resource, or call agenthold_wait to be notified when it becomes available."
}

已声明（你已持有此声明，幂等）：

{ "status": "already_claimed", "resource": "file://default/intro.md", "version": 1 }

---

agenthold_release

释放你的声明，并明确说明你所做的操作。此结果会保存在空闲状态记录中，并显示给下一个声明者，避免其依据过时的假设采取行动。需要已注册的 agent_id。

{
  "resource": "intro.md",
  "agent_id": "agent-a1b2c3d4",
  "outcome": "modified"
}

{ "status": "released", "resource": "file://default/intro.md", "version": 2, "outcome": "modified" }

结果：released（默认值——不声明生命周期状态）、modified（已就地修改）、created（之前不存在）、deleted（此资源位置已不存在）、moved（已移动——还需传入新资源的 moved_to）。

重命名时（mv old new）：声明两个路径，在磁盘上执行重命名，然后以 outcome: "moved", moved_to: "new" 释放源路径，并以 outcome: "created" 释放目标路径。

---

agenthold_status

检查资源是否可用或当前是否已被声明。无需注册。

{ "resource": "intro.md" }

可用：

{ "status": "available", "resource": "file://default/intro.md" }

如果前一个持有者声明了非简单的结果（deleted、moved、abandoned、expired），响应还会包含 previous_outcome、previous_holder、previous_outcome_at 和 hint。当结果为 moved 时，moved_to 是新的资源 URI。

已声明：

{
  "status": "claimed",
  "resource": "file://default/intro.md",
  "held_by": "agent-e5f6g7h8",
  "agent_name": "editor-agent",
  "agent_model": "claude-sonnet-4-6",
  "claimed_at": "2026-03-18T10:00:00+00:00",
  "version": 3
}

---

agenthold_wait

等待已声明的资源变为可用。在持有者释放资源或超时之前，会阻塞 Agent 回合。

{ "resource": "intro.md", "timeout_seconds": 30 }

可用（资源已释放）：

{ "status": "available", "resource": "file://default/intro.md", "elapsed_seconds": 2.4 }

如果等待因非简单结果的释放而结束，响应还会包含 previous_outcome、previous_holder、previous_outcome_at、可选的 moved_to 和 hint。等待编辑已移动或已删除资源的 Agent 可以了解路径变为可用的原因。

超时：

{
  "status": "timeout",
  "resource": "file://default/intro.md",
  "held_by": "writer-2",
  "elapsed_seconds": 30.2,
  "hint": "The resource was not released within the timeout. Try working on a different resource, or call agenthold_wait again with a longer timeout."
}

---

高级工具

对于自定义协调协议，agenthold 通过 --tools advanced 提供八个底层原语：

agenthold_get · agenthold_set · agenthold_list · agenthold_history · agenthold_delete · agenthold_watch · agenthold_clear_namespace · agenthold_export

这些工具让 Agent 能够直接读取、写入和监视带版本的状态存储，并进行完整的 OCC 冲突检测。此模式下不会发送服务器指令。

查看完整的高级工具参考 →

---

冲突检测

使用 expected_version 的读-改-写模式是 agenthold 的核心。以下是标准重试循环：

from agenthold.store import StateStore
from agenthold.exceptions import ConflictError

store = StateStore("./state.db")

record = store.get("campaign", "budget")   # read once before doing work
do_expensive_work()                         # LLM call, API request, etc.

while True:
    new_value = compute_new_value(record.value)
    try:
        store.set(
            "campaign", "budget", new_value,
            updated_by="my-agent",
            expected_version=record.version,
        )
        break   # write succeeded
    except ConflictError:
        record = store.get("campaign", "budget")   # re-read and retry

**为什么有效：**版本号就是契约。如果你读取后存储的版本已递增，说明另一个 Agent 先写入了。你获取当前值，重新计算，然后重试。重试次数受并发写入者数量限制。实际上，Agent 几乎不会发生超过一次的冲突。

**为什么不使用锁？**锁需要租约机制（如果 Agent 持锁崩溃会怎样？），每次读取都会增加延迟，并且难以应对 Agent 工作流中固有的长时间 I/O 等待。OCC 仅在实际发生冲突时才会产生开销。

---

用作 Python 库

from agenthold.store import StateStore
from agenthold.exceptions import ConflictError

store = StateStore("./state.db")

# Write a value (first write, no conflict check needed)
store.set("order-1234", "status", "received", updated_by="intake-agent")

# Read it back; always get the version number too
record = store.get("order-1234", "status")
print(record.value)    # "received"
print(record.version)  # 1

# Write with conflict detection; pass the version you read
try:
    store.set(
        "order-1234", "status", "processing",
        updated_by="fulfillment-agent",
        expected_version=record.version,  # rejected if another agent wrote first
    )
except ConflictError as e:
    # Another agent wrote between your read and write.
    # e.detail has the current version, value, and who wrote it.
    record = store.get("order-1234", "status")
    # ... recalculate and retry

---

实际效果

在多 Agent 会话中，协调会自动完成。Agent 的工具调用如下所示：

Agent A: agenthold_register(name="writer", model="claude-sonnet-4-6")
         → agent_id: "agent-a1b2c3d4"

Agent A: agenthold_claim(resource="chapter-3.md", agent_id="agent-a1b2c3d4")
         → status: "claimed", resource: "file://default/chapter-3.md"

Agent B: agenthold_claim(resource="chapter-3.md", agent_id="agent-e5f6g7h8")
         → status: "busy", hint: "Work on a different resource..."

Agent A: agenthold_release(resource="chapter-3.md", agent_id="agent-a1b2c3d4", outcome="modified")
         → status: "released", outcome: "modified"

无需设计系统提示词。工具描述会引导 Agent。

示例演示

其中包含两个示例，每个示例都有一个“前置”和“后置”脚本。

订单处理：两个 Agent 并发更新同一条订单记录：

uv run python examples/order_processing/without_agenthold.py   # silent overwrite
uv run python examples/order_processing/with_agenthold.py      # conflict detection + retry

预算分配：两个 Agent 从共享的营销预算中支取资金：

uv run python examples/budget_allocation/without_agenthold.py  # $10k budget → $15k committed
uv run python examples/budget_allocation/with_agenthold.py     # exact allocation, full audit trail

---

配置

agenthold --db ./state.db                              # standard mode (default)
agenthold --db ./state.db --tools advanced             # advanced mode
agenthold --db ./state.db --claim-ttl 1800             # standard + 30 min TTL
agenthold --workspace myproj=/abs/path                 # named workspace
agenthold --workspace a=/x --workspace b=/y            # multiple workspaces
agenthold --transport http --port 8417                 # serve over HTTP (see below)

标志	默认值	说明
--db	./agenthold.db	SQLite 数据库文件的路径。使用 :memory: 可创建进程内存储（仅用于测试；进程退出时数据会丢失）。
--tools	standard	工具集：standard（register/claim/release/status/wait）或 advanced（get/set/delete/watch/list/history/clear/export）。
--claim-ttl	None (no expiry)	非活跃 Agent 的声明在多少秒后过期。仅适用于 standard 模式。设置此项后，最后活动时间超过此值的 Agent 所持有的声明将被视为已过期，其他 Agent 可以接管。建议与 --transport http 配合使用。
--workspace	当前工作目录中名为 default 的工作区	将工作区配置为 name=path（例如 myproj=/abs/path）或绝对路径（名称根据目录名确定）。可重复指定。多个工作区可让一个 agenthold 进程通过同一数据库协调多个独立代码库。工具输入中的裸路径会相对于名为 default 的工作区解析；如果只配置了一个工作区，则相对于该工作区解析。
--transport	stdio	stdio（每个 Agent 启动一个本地子进程）或 http（启动一个长期运行的服务，供多个 Agent 通过 Streamable HTTP 连接）。
--host	127.0.0.1	HTTP 绑定地址。仅与 --transport http 配合使用。
--port	8417	HTTP 端口。仅与 --transport http 配合使用。
--path	/mcp	MCP 传输所挂载的 HTTP 端点路径。仅与 --transport http 配合使用。
--json-response	off	通过 HTTP 返回 JSON 响应，而非 SSE 流。仅与 --transport http 配合使用。
--allowed-host	none	启用 DNS 重绑定保护，并允许此 Host 标头值。可重复指定。省略时，保护保持禁用状态（便于 localhost 使用）。仅与 --transport http 配合使用。
--auth-token	none	要求每个 HTTP 请求都携带此 bearer 令牌（Authorization: Bearer <token>）。可重复指定。也可通过 AGENTHOLD_AUTH_TOKEN 环境变量设置（以逗号分隔），并且更推荐使用环境变量。仅与 --transport http 配合使用。

数据库文件会在首次运行时自动创建。像备份其他 SQLite 文件一样备份它。

HTTP 传输

默认情况下，agenthold 使用 stdio：MCP 客户端为每个 Agent 启动一个 agenthold 子进程，这些子进程通过同一台机器上的共享 SQLite 文件进行协调。这非常适合在单台机器上运行本地多 Agent 任务。

要协调运行在不同进程、容器或主机中的 Agent，请通过 Streamable HTTP 运行一个长期服务，并让每个 Agent 都连接到该服务：

agenthold --db ./state.db --transport http --host 127.0.0.1 --port 8417 --claim-ttl 1800

然后通过 URL 而非命令连接 MCP 客户端：

{
  "mcpServers": {
    "agenthold": {
      "url": "http://127.0.0.1:8417/mcp"
    }
  }
}

全部五种标准工具（以及使用 --tools advanced 时的高级工具集）在 HTTP 上的工作方式完全相同——变化的只有传输方式。由于现在一个服务会在其运行期间为多个 Agent 提供服务，请设置 --claim-ttl，这样断开连接的 Agent 所持有的声明就能自动被重新接管。

身份验证

若要对每个 HTTP 请求都要求携带 bearer token，请设置一个或多个 token。建议使用环境变量，避免 token 出现在进程列表中：

AGENTHOLD_AUTH_TOKEN=s3cr3t agenthold --db ./state.db --transport http --host 0.0.0.0
# or, less privately:
agenthold --transport http --auth-token s3cr3t --auth-token another-token

随后，客户端在 Authorization 标头中发送 token：

{
  "mcpServers": {
    "agenthold": {
      "url": "http://your-host:8417/mcp",
      "headers": { "Authorization": "Bearer s3cr3t" }
    }
  }
}

没有有效 token 的请求会在到达存储之前被拒绝，并返回 401。Token 会以恒定时间进行比较。身份验证仅适用于 --transport http（stdio 是可信的本地传输方式）。此版本的身份验证要么全部启用，要么全部不启用；按命名空间进行范围限定已列入路线图。

AGENTHOLD_AUTH_TOKEN 的值会按逗号拆分，因此 token 本身不能包含逗号（对于此类 token，请重复使用 --auth-token）。如果请求启用身份验证，但所有 token 都是空白，服务会拒绝启动，而不是在没有身份验证的情况下运行。

安全。Bearer token 的安全性取决于传输它的方式——在任何实际部署中，请在 agenthold 前面配置 TLS（反向代理）。 默认绑定地址为 127.0.0.1（仅限 localhost）；若绑定到 localhost 之外的地址且未设置 --auth-token，启动时会打印警告。传入 --allowed-host <host>（可重复）以启用 DNS 重绑定（Host 标头）防护——请注意，这不是身份验证，也不能替代 --auth-token。

**长期运行的服务。**agenthold 目前还不会清理闲置的 Agent 记录，因此，若服务长时间运行且 Agent 频繁更替，Agent 元数据会不断累积。设置 --claim-ttl（以便恢复已断开连接的 Agent 所持有的声明），并在需要时定期重启。自动清理已列入路线图。

---

检查存储

agenthold 数据库只是一个 SQLite 文件，但你通常不需要手动打开它。agenthold 命令也可用作只读检查器，让你一眼了解各 Agent 的活动——无需 MCP 客户端。这些命令严格只读（数据库以 query_only 模式打开，且不会写入架构），因此绝不会修改存储——即使你将 --db 指向了错误的文件。

agenthold agents --db ./state.db          # who is registered
agenthold claims --db ./state.db          # what is currently claimed, and by whom
agenthold claims --db ./state.db --all    # also show freed/moved/deleted claims
agenthold namespaces --db ./state.db      # namespaces + record counts
agenthold keys orders --db ./state.db     # keys in a namespace
agenthold history orders order-1 --db ./state.db   # a key's version history

示例：

$ agenthold claims --db ./state.db
RESOURCE                    STATE    BY            SINCE
--------------------------  -------  ------------  ------
custom://chapter-1          claimed  editor-agent  2s ago
file://default/src/main.py  claimed  review-bot    2s ago

每个命令都接受 --db PATH（默认值为 ./agenthold.db）和 --json，用于输出机器可读内容。直接运行 agenthold（不带子命令）仍会像以前一样启动 MCP 服务——检查器只是在原有功能上的补充。

---

开发

git clone https://github.com/edobusy/agenthold.git
cd agenthold
uv sync --all-extras --dev

运行测试：

uv run pytest tests/ -v

检查覆盖率：

uv run pytest tests/ --cov=agenthold --cov-report=term-missing

Lint 和类型检查：

uv run ruff check src/ tests/
uv run ruff format src/ tests/
uv run mypy src/

CI 会在每次推送到 main 时使用 Python 3.11 和 3.12 运行。详细指南请参阅 CONTRIBUTING.md。

---

技术说明（面向工程师的设计决策）

为什么使用 SQLite？ 对于当前范围，SQLite 是合适的工具。它无需依赖，包含在 Python 标准库中，并且可以在任何环境中运行。已启用 WAL 模式，因此只读操作（导出、监视）不会阻塞跨进程写入。写事务使用 BEGIN IMMEDIATE 提前获取写锁，确保即使多个 agenthold 进程共享同一个数据库文件，OCC 冲突检测也能正常工作。busy_timeout 设置为 5 秒，因此第二个写入方会等待，而不是立即失败。Postgres 在这个规模下没有任何收益，只会增加运维依赖。存储后端通过简洁的接口（StateStore）封装，必要时可以替换为 Postgres。刻意选择简单的工具并不意味着它存在局限。

为什么选择 OCC，而不是悲观锁？ 锁需要持有者释放，这意味着系统必须处理崩溃、超时和过期锁持有者。当冲突很少时，这种复杂性并不值得。OCC 仅在实际发生冲突时才会产生开销：额外读取一次并重试一次。对于多 Agent 工作流，Agent 在读写之间会执行大量工作（大模型推理、API 调用、工具执行），OCC 是正确的选择。

版本控制保证： 每个键都有一个从 1 开始的版本，每次写入时版本恰好递增 1。state_history 表仅追加记录，并且会在更新实时记录之前记录每次写入，因此即使两次写入之间发生崩溃，历史记录仍保持一致。删除操作也会先向 state_history 写入一条墓碑记录（其中 event_type: "delete"），然后再移除实时记录，因此可以在历史记录中查看键的完整生命周期。顺序保证针对每个键，而非全局；两个不同键的版本可以按任意顺序更新。

生产规模下需要做哪些改变： 主要只有一项：将 SQLite 替换为 Postgres，以获得更高的并发写入吞吐量、复制能力和托管服务。StateStore 接口的设计已确保这项改动范围可控。网络传输和身份验证功能已经具备——agenthold 通过 MCP SDK 的 StreamableHTTPSessionManager 提供 Streamable HTTP（因此远程 Agent 可通过网络连接，而非使用本地进程），并通过 --auth-token 使用 bearer token 控制访问（参见身份验证）。生产环境中应在前面配置 TLS 反向代理；请注意，目前尚未实现按命名空间/租户划分令牌的功能。

---

许可证

MIT。请参阅 LICENSE。

---

mcp-name: io.github.edobusy/agenthold