幂等键：把重复请求认作同一次操作

mkdir -Force "F:\学习\agentdepth-l2-idempotency"
cd "F:\学习\agentdepth-l2-idempotency"

"""用内存笔记观察参数校验、资源归属和重复写入。"""

from copy import deepcopy


class ToolError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


class NoteStore:
    def __init__(self):
        # 两位用户各有一篇笔记，owner 保存归属。
        self.notes = {
            "n1": {"owner": "u1", "text": "仅供 u1 读取的示例笔记"},
            "n2": {"owner": "u2", "text": "u2 的示例笔记"},
        }
        # reads 记录成功读取次数，effects 记录实际创建次数。
        self.reads = 0
        self.effects = 0
        self.seen_keys = {}

    def read(self, identity, note_id):
        note = self.notes.get(note_id)
        if note is None or note["owner"] != identity:
            raise ToolError("forbidden", "resource_not_owned")
        self.reads += 1
        return {"note_id": note_id, "text": note["text"]}

    def create_tag(self, identity, note_id, tag, idem_key):
        if identity != "u1" or note_id != "n1":
            raise ToolError("forbidden", "resource_not_owned")
        # 同一个业务键对应同一次创建动作，保存输入和结果用于去重。
        payload = {"identity": identity, "note_id": note_id, "tag": tag}
        if idem_key in self.seen_keys:
            if self.seen_keys[idem_key]["payload"] != payload:
                raise ToolError("idempotency_conflict", "same_key_different_payload")
            return deepcopy(self.seen_keys[idem_key]["result"])
        self.effects += 1
        result = {"tag_id": "tag-%d" % self.effects, "status": "created"}
        self.seen_keys[idem_key] = {"payload": deepcopy(payload), "result": result}
        return deepcopy(result)


def call_tool(request, identity, store):
    """依次检查工具名、输入字段、资源归属及返回值。"""
    if not isinstance(request, dict):
        raise ToolError("invalid_request", "object_required")
    if request.get("name") not in ("read_note", "create_tag"):
        raise ToolError("unknown_tool", "tool_not_allowlisted")
    args = request.get("arguments")
    if not isinstance(args, dict):
        raise ToolError("invalid_arguments", "object_required")
    if request["name"] == "read_note":
        if set(args) != {"note_id"} or not isinstance(args["note_id"], str) or not args["note_id"].strip():
            raise ToolError("invalid_arguments", "read_note_schema")
        # 先核对笔记归属，再执行读取；读取成功后计数增加。
        note = store.notes.get(args["note_id"])
        if note is None or note["owner"] != identity:
            raise ToolError("forbidden", "resource_not_owned")
        result = store.read(identity, args["note_id"])
        if (not isinstance(result, dict) or set(result) != {"note_id", "text"}
                or not isinstance(result["note_id"], str) or not result["note_id"]
                or not isinstance(result["text"], str)):
            raise ToolError("invalid_tool_result", "read_note_output_schema")
        return result
    if set(args) != {"note_id", "tag", "idempotency_key"}:
        raise ToolError("invalid_arguments", "create_tag_schema")
    if not all(isinstance(args[k], str) and args[k].strip() for k in args):
        raise ToolError("invalid_arguments", "non_empty_strings_required")
    return store.create_tag(identity, args["note_id"], args["tag"], args["idempotency_key"])


def demo():
    store = NoteStore()
    cases = [
        ("own_read", {"name": "read_note", "arguments": {"note_id": "n1"}}, "u1"),
        ("cross_user_read", {"name": "read_note", "arguments": {"note_id": "n1"}}, "u2"),
        ("unknown_tool", {"name": "delete_all", "arguments": {}}, "u1"),
    ]
    # 收集每个请求的结果，最后查看实际读取和创建的次数。
    result = []
    for label, request, identity in cases:
        try:
            value = call_tool(request, identity, store)
            result.append({"case": label, "status": "ok", "value": value})
        except ToolError as error:
            result.append({"case": label, "status": "error", "code": error.code})
    return {"cases": result, "body_reads": store.reads, "effects": store.effects}


if __name__ == "__main__":
    print(demo())

"""重复提交同一次添加标签请求，观察返回值与创建次数。"""
from tool_gate import NoteStore, ToolError, call_tool
import json

store = NoteStore()
request = {"name": "create_tag", "arguments": {
    "note_id": "n1", "tag": "todo", "idempotency_key": "op-1"
}}
# 第一次创建，第二次取得同一次操作保存的结果。
for _ in range(2):
    print(json.dumps(call_tool(request, "u1", store)))
print("effects", store.effects)
# 同一个操作键换了内容，会返回冲突。
request["arguments"]["tag"] = "urgent"
try:
    call_tool(request, "u1", store)
except ToolError as error:
    print("conflict", error.code)

python tag_demo.py

{"tag_id": "tag-1", "status": "created"}
{"tag_id": "tag-1", "status": "created"}
effects 1
conflict idempotency_conflict

for key in ("op-1", "op-2"):
    request["arguments"]["idempotency_key"] = key
    print(json.dumps(call_tool(request, "u1", store)))

python -m unittest test_tool_gate -v