CMA：MongoDB Atlas 集成

使用 MongoDB Atlas 作为 Agent 后端

🌐 查看英文原文 | 源码 Notebook

基于 MongoDB Atlas 和 Claude 管理代理的欺诈审查代理

本食谱展示了如何仅使用标准 CMA 模式（自定义工具和 MCP 工具集），将 MongoDB Atlas 带到运行在 Claude 管理代理（CMA）上的代理中——作为其检索引擎、其图存储和其记录系统——无需平台级 MongoDB 集成。可运行的助手位于此笔记本旁边的 mongodb_on_cma/（mongodb_on_cma/）。每个部分都链接到完整文件，以便叙事始终集中在 MongoDB 的接入点 和 代理增加的价值 上。

MongoDB 的接入点： 大多数代理堆栈将三个或四个系统连接在一起：一个用于语义搜索的向量数据库、一个用于关键词的独立引擎、一个用于关系的图存储以及一个用于记录本身的操作数据库。每个缝隙都是另一个集成、另一个凭证、另一个代理对世界看法可能偏离的地方。MongoDB 将这些压缩成一个引擎：相同的文档可以通过向量 ($vectorSearch)、通过全文 ($search)、作为两种融合的混合（与相互排名融合，即 RRF ($rankFusion)）进行搜索，并且可以作为图 ($graphLookup) 进行遍历——并且它们是代理读取、写入和持久化其决策的相同文档。一种查询语言、一个集群、一个连接 (pymongo）。

代理增加的价值： CMA 在 Anthropic 的一侧运行代理循环和沙盒。您的应用程序拥有 MongoDB 的 数据路径。代理在 MongoDB 检索的内容上进行推理，通过 CMA 的原生 requires_action 网关暂停对有风险的调用，并将其裁决写回——所有这些都不需要数据库凭证进入代理上下文或其沙盒。 > MongoDB 这里是你的 操作数据存储和检索引擎 — 你的记录系统，由你的应用程序控制。这与CMA在平台上原生管理的内存功能是不同的。两者是互补的。

到结束时，你将能够：

以三种方式将凭证安全的MongoDB Atlas数据路径连接到Claude Managed Agent。
将向量、全文、混合和图检索提升到代理的定制工具中。
在CMA的原生人工介入暂停中屏蔽有风险的代理决策。
将一个MongoDB Atlas集群作为代理的记录系统和审计骨干。

这个示例是一个 人工介入的欺诈审查代理，但模式是垂直无关的：交换集合和工具，相同的形状可以服务于支持、研究或运营代理。

本指南涵盖的内容

将MongoDB连接到管理代理
检索：四种模式，一个引擎
端到端代理：人工介入的欺诈审查
MongoDB Atlas作为记录系统和审计骨干
总结python %%capture # Third-party dependencies: the Anthropic SDK (drives Claude Managed Agents via client.beta.*), # pymongo (the host-side data path), python-dotenv, and cryptography + pyjwt (sign/verify the # AP2 mandates). This installs the PyPI packages only — the notebook also imports the adjacent # `mongodb_on_cma/` package and `utilities.py`, so run it from a clone of this repo (see Setup). %pip install -q "anthropic>=0.109.0" pymongo python-dotenv cryptography pyjwt## 前提条件

所需知识： Python 和 pymongo 基础知识，以及 Claude API 的基本了解。MongoDB + Claude 新手？RAG notebook 库首先介绍了 $vectorSearch。自定义工具门控新手？CMA_gate_human_in_the_loop.ipynb 介绍了往返过程。

所需工具： Python 3.11+，Anthropic API 密钥，以及 MongoDB Atlas 集群。

设置

所需： MONGO_URI（一个 Atlas SRV 连接字符串）以及 Anthropic API 访问权限。免费 M0 集群运行本食谱中的所有内容—— 向量、全文、混合 $rankFusion 和图遍历——因此没有付费层级的必要。混合搜索使用本地的 $rankFusion 阶段，需要 MongoDB 8.0+；每个当前的 Atlas 集群（包括 M0）都在 8.0 或更高版本，因此默认情况下即可满足要求。bash uv sync --all-extras # from the repo root cp .env.example .env # then add MONGO_URI (and Anthropic auth, if not already configured)可选的嵌入/重排提供者： 种子配置文件已预计算嵌入，因此烹饪书无需嵌入即可运行。要启用实时嵌入和重排路径，请设置 MDB_ATLAS_API_KEY（MongoDB Atlas AI端点，提供 /v1/embeddings 和 /v1/rerank）或 VOYAGE_API_KEY（voyageai SDK）。ENABLE_RERANK=1 添加重排的第二阶段。AUTO_APPROVE=1 以确定性方式解决CI的人门关卡。COOKBOOK_MODEL 覆盖代理模型（默认 claude-haiku-4-5）。

还没有Atlas集群？ 合作笔记本 rag_using_mongodb.ipynb 演示了创建免费集群并获取您的 MONGO_URI；Atlas Search索引文档覆盖索引设置。

教学代码位于 此笔记本中行内：四个检索管道构建器（第2节）和自定义工具处理程序以及 requires_action 网关循环（第3节）。旁边存放的 mongodb_on_cma/ 包的笔记本仅包含设置模板，您导入而不是读取：config.py（索引名称 + 可调参数）， embeddings.py（Atlas/Voyage 客户端）， tools.py（Atlas 种子/索引设置 + 共享决策/审计文档 shapers），ap2_mandates.py（AP2 签名/验证 — 一个通过 verify_mandates 调用的加密黑盒），以及 seed.py（示例数据加载器）。导入它们，连接客户端，并解决运行配置。```python import hashlib import json import logging import os from datetime import UTC, datetime from typing import Any

import dotenv from anthropic import Anthropic from mongodb_on_cma import ( EMBED_DIM, build_audit_event, build_decision_doc, ensure_indexes, has_anthropic_auth, load_seed, make_embedding_client, missing_required_env, preflight, prepare_seed, rerank, resolve_model, seed_collection, server_version, supports_rank_fusion, ) from mongodb_on_cma.ap2_mandates import ( attach_mandates, store_mandate_receipt, tool_verify_mandates, ) from pymongo import MongoClient from utilities import wait_for_idle_status

dotenv.load_dotenv()

missing = missing_required_env() assert not missing, f”Set these in your environment / .env: {missing}” # Anthropic auth is a soft check: an API key, an auth token, or a profile (e.g. the ant # CLI’s workload-identity federation) all work, so warn rather than block if none is set. if not has_anthropic_auth(): print( “warning: no ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN / ANTHROPIC_PROFILE found;” “relying on the SDK / ant CLI to resolve credentials.” )

MODEL = resolve_model() ENABLE_RERANK = os.getenv(“ENABLE_RERANK”, ““).lower() in (”1”, “true”) AUTO_APPROVE = os.getenv(“AUTO_APPROVE”, ““).lower() in (”1”, “true”)

Quiet the SDK’s one-shot notice that ANTHROPIC_API_KEY shadows profile/federation

auto-discovery — expected when an API key is set; harmless under `ant` / WIF auth.

logging.getLogger(“anthropic.lib.credentials._auth”).setLevel(logging.ERROR) client = Anthropic() mongo = MongoClient(os.environ[“MONGO_URI”]) db = mongo[“fraud_review_demo”] coll = db[“transactions”] ai_client = make_embedding_client() # Atlas / Voyage / None (seed ships precomputed vectors)

print( f”model={MODEL} rerank={‘on’ if ENABLE_RERANK else ‘off’} provider={‘yes’ if ai_client else ‘none’}” )

warning: no ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN / ANTHROPIC_PROFILE found; relying on the SDK / ant CLI to resolve credentials. model=claude-haiku-4-5 rerank=on provider=yes


## 1. 将 MongoDB 连接到托管代理

代理循环及其沙盒是 **Anthropic 主办的** — 这就是“托管”的含义。你唯一需要托管的是 **数据路径** 到 MongoDB。在每个路径上，MongoDB 凭证
都位于 *你的* 边界一侧：它永远不会进入代理上下文、云沙盒的环境或代理可以读取的文件。
> **为什么这个食谱通过 `client.beta.*` 与 CMA 通信，而不是 Claude 代理 SDK。**
> 托管代理（CMA）是一个 [托管 REST API](https://platform.claude.com/docs/en/managed-agents/overview)：

> Anthropic 运行代理循环和沙盒，而你通过创建代理/会话
> 并通过标准 Anthropic SDK 的 beta 接口流式传输事件来驱动它（`client.beta.agents`，`client.beta.sessions`，`client.beta.sessions.events`）。没有
> 独立的“CMA SDK” — 这就是使用 CMA 的方法。The
> [Claude 代理 SDK](https://code.claude.com/docs/en/agent-sdk/overview)
> (`client.beta.agents`, `client.beta.sessions`, `client.beta.sessions.events`). There is no
> separate "CMA SDK" — this *is* how you use CMA. The
> [Claude Agent SDK](https://code.claude.com/docs/en/agent-sdk/overview)
> (`claude-agent-sdk`，在仓库中展示）是一个不同的产品：一个在您自己的进程、您自己的基础设施上运行代理循环的库。
> [`claude_agent_sdk/`](../claude_agent_sdk/) 烹饪书（cookbooks））是：一个库，它在您的进程中运行代理循环。
> 这两个产品是互补的——您将使用代理SDK进行本地原型设计，并在CMA上运行生产环境——但这样的CMA集成是基于beta会话/事件API构建的。这个选择也是下面凭证边界工作的原因：因为 *您的* 进程处理每个自定义工具调用，MongoDB的秘密就保持在主机端（见路径A）。
> 有三种方式来连接数据路径。**这个烹饪书在整个过程中使用路径A**——这是推荐的、最轻量级的模式，并且是保持凭证完全在主机端的模式。路径B和C在后面总结，以保持完整性。
> ### 路径A：主机端自定义工具（推荐）
> 您的应用程序定义了一个 `custom` 工具。当代理调用它时，会话会暂停，显示 `requires_action`，**您的** 进程使用 `pymongo` 运行查询，然后您发送回一个 `user.custom_tool_result`。秘密永远不会进入沙盒，MongoDB可以放在一个私有的VPC中，代理被限制在固定、经过审计的查询集。这是最轻量级的路径——没有持续的服务器——并且是Anthropic为任何携带秘密数据源推荐的模式。
> 为了清晰地看到连接，这里有一个完整的往返过程，使用一个微小的 `find_notes` 工具在一个废弃的 `notes` 集合上。**这是核心集成：几行 `pymongo` 代码，在自定义工具后面以主机端方式运行。** 第3节将这种精确的形状扩展到五个工具和一个真实的

> 打开事件流，提出问题，并驱动往返：在每个
the recommended, lightest pattern and the one that keeps the credential fully host-side. Paths
B and C are summarized after it for completeness.

### Path A: host-side custom tool (recommended)

Your application defines a `custom` tool. When the agent calls it, the session pauses with
`requires_action`, **your** process runs the query with `pymongo`, and you post back a
`user.custom_tool_result`. The secret never enters the sandbox, MongoDB can sit in a private
VPC, and the agent is constrained to a fixed, audited set of queries. It is the lightest path —
no standing server — and Anthropic's recommended pattern for any secret-bearing data source.

To see the wiring unobscured, here is the whole round-trip with one tiny `find_notes` tool over
a throwaway `notes` collection. **This is the core integration: a few lines of `pymongo`, run
host-side, behind a custom tool.** Section 3 scales this exact shape to five tools and a real
workload.

```python
notes = mongo["cookbook_mongodb_on_cma"]["notes"]
notes.delete_many({})
notes.insert_many(
    [
        {"note_id": "n1", "tag": "todo", "text": "Renew the SSL certificate before it expires."},
        {"note_id": "n2", "tag": "idea", "text": "Add a dark-mode toggle to the dashboard."},
        {"note_id": "n3", "tag": "todo", "text": "Email the quarterly report to the finance team."},
    ]
)

notes_agent = client.beta.agents.create(
    name="cookbook-mongodb-notes",
    model=MODEL,
    system=(
        "You answer questions about the user's notes. Use the find_notes tool to fetch notes "
        "(optionally filtered by `tag`) before answering, then give a concise final answer."
    ),
    tools=[
        {
            "type": "agent_toolset_20260401",
            "default_config": {"enabled": True, "permission_policy": {"type": "always_allow"}},
        },
        {
            "type": "custom",
            "name": "find_notes",
            "description": "Return stored notes, optionally filtered by tag (e.g. 'todo', 'idea').",
            "input_schema": {
                "type": "object",
                "properties": {"tag": {"type": "string", "description": "optional tag filter"}},
                "required": [],
            },
        },
    ],
)
notes_env = client.beta.environments.create(
    name="cookbook-mongodb-notes-env", config={"type": "cloud", "networking": {"type": "limited"}}
)
notes_session = client.beta.sessions.create(
    environment_id=notes_env.id,
    agent={"type": "agent", "id": notes_agent.id, "version": notes_agent.version},
    title="MongoDB notes (host-side custom tool)",
)
print(f"seeded {notes.count_documents({})} notes; session: {notes_session.id}")

seeded 3 notes; session: sesn_016ufm58BVJ8aZeWjUt39bsq

Open the event stream, ask a question, and drive the round-trip: on each agent.custom_tool_use，使用pymongo在主机端运行查询，并将结果user.custom_tool_result反馈回来。结果使用{"_id": 0}进行投影，因为原始MongoDB ObjectId无法进行JSON序列化。路径A是默认路径，本菜谱的其他部分都使用它。存在两种替代方案，在不适用的情况下——您不需要它们来跟随：

def run_find_notes(args: dict) -> dict:
    """Host-side handler: pymongo runs here; _id (ObjectId) is projected out so the result is JSON-able."""
    flt = {"tag": args["tag"]} if args.get("tag") else {}
    return {"notes": list(notes.find(flt, {"_id": 0}))}


tool_use_events, responded, round_trips = {}, set(), []
question = "Using find_notes, list every note tagged 'todo', then tell me how many there are."

with client.beta.sessions.events.stream(notes_session.id) as stream:
    client.beta.sessions.events.send(
        session_id=notes_session.id,
        events=[{"type": "user.message", "content": [{"type": "text", "text": question}]}],
    )
    for ev in stream:
        if ev.type == "agent.custom_tool_use":
            tool_use_events[ev.id] = ev
        elif ev.type == "session.status_idle" and ev.stop_reason:
            if ev.stop_reason.type == "requires_action":
                for event_id in ev.stop_reason.event_ids:
                    if event_id in responded:
                        continue
                    tev = tool_use_events[event_id]
                    result = (
                        run_find_notes(tev.input)
                        if tev.name == "find_notes"
                        else {"error": "unknown"}
                    )
                    if tev.name == "find_notes":
                        round_trips.append(
                            {"tool_input": tev.input, "returned": len(result["notes"])}
                        )
                    client.beta.sessions.events.send(
                        session_id=notes_session.id,
                        events=[
                            {
                                "type": "user.custom_tool_result",
                                "custom_tool_use_id": event_id,
                                "content": [{"type": "text", "text": json.dumps(result)}],
                            }
                        ],
                    )
                    responded.add(event_id)
            elif ev.stop_reason.type == "end_turn":
                break
        elif ev.type == "session.status_terminated":
            break

wait_for_idle_status(client, notes_session.id)
print("host-side find_notes round-trips:")
for rt in round_trips:
    print(f"  find_notes({rt['tool_input']}) -> {rt['returned']} notes")

client.beta.sessions.archive(notes_session.id)
client.beta.environments.archive(notes_env.id)
client.beta.agents.archive(notes_agent.id)
notes.drop()
print("archived the demo agent / environment / session; dropped the notes collection")

host-side find_notes round-trips:
  find_notes({'tag': 'todo'}) -> 2 notes
archived the demo agent / environment / session; dropped the notes collection

Other options (brief)

路径B——沙箱内的pymongo（自托管沙箱）。 如果您在自己的基础设施上运行沙箱，MongoDB客户端可以位于沙箱内部，并且代理可以直接从其bash工具查询MongoDB，使用MONGO_URI作为普通的容器环境变量。仅适用于自托管沙箱（在云沙箱上，连接字符串将出现在会话历史中）。可运行镜像：self_hosted_sandboxes/docker/。
路径C——自托管MongoDB MCP服务器。 在HTTPS后托管官方MongoDB MCP服务器，并将其注册为mcp_toolset，以便代理可以查询整个表面（find、aggregate、$vectorSearch），而不是固定的查询集。标准CMA MCP + vault连接。一句话的权衡：路径A保持查询集固定，并将秘密存储在您的后端（最安全的默认设置）；路径B将沙箱直接访问权交给；路径C以运行服务器为代价，给代理提供最广泛的查询表面。如有疑问，请保持在路径A。 ## 2. 检索：四种模式，一个引擎代理的根基在于其检索能力。这些是独立显示的四个MongoDB检索模式——这里有构建器，这里有它返回的内容——因此您可以将单个模式提升到自己的代理工具中。每个都是一个普通的聚合管道构建器；每个都 Runnable image: self_hosted_sandboxes/docker/.
Path C — self-hosted MongoDB MCP server. Host the official MongoDB MCP server behind HTTPS and register it as an mcp_toolset so the agent can query the full surface (find, aggregate, $vectorSearch) rather than a fixed query set. Standard CMA MCP + vault wiring.

The trade-off in one line: Path A keeps the query set fixed and the secret in your backend (the safest default); Path B hands the sandbox direct access; Path C gives the agent the broadest query surface at the cost of running a server. When in doubt, stay on Path A.

2. Retrieve: four patterns, one engine

An agent is only as well-grounded as its retrieval. These are the four MongoDB retrieval patterns shown in isolation — here’s the builder, here’s what it returns — so you can lift a single pattern into your own agent’s tools. Each is a plain aggregation-pipeline builder; the 第3节中，Agent通过其工具调用这些相同的函数。

模式	构建器	MongoDB阶段
向量搜索	`build_vector_pipeline`	`$vectorSearch`
全文搜索	`build_lexical_pipeline`	`$search`
混合（互逆排名融合）	`build_rank_fusion_pipeline`	`$rankFusion` (8.0+)
图遍历	`build_graph_pipeline`	`$graphLookup`

构建器在下一单元格中定义（内联）——每个构建器都返回一个聚合阶段列表，您可以将其提升到自己的集合中。索引名称和投影字段来自 config.py（mongodb_on_cma/config.py）；EMBED_DIM和索引常量在首先从 setup cell above.

example_data/mongodb_on_cma/（example_data/mongodb_on_cma/seed_transactions.jsonl）加载种子固定装置并构建索引。该固定装置包含预计算的嵌入，因此无需提供者密钥即可运行。 ### 模式1：向量搜索（$vectorSearch）

# The four retrieval builders — each returns a plain aggregation pipeline you can lift into your
# own collection. These are the functions Section 3's agent calls through its custom tools.
from mongodb_on_cma.config import (
    DECIDED_STATUSES,
    LEXICAL_PATHS,
    PROJECT_FIELDS,
    SEARCH_INDEX_NAME,
    VECTOR_INDEX_NAME,
)


def _project_stage(with_score: bool = False) -> dict:
    proj: dict[str, Any] = {f: 1 for f in PROJECT_FIELDS}
    proj["_id"] = 0
    if with_score:
        proj["score"] = {"$meta": "score"}
    return {"$project": proj}


def build_vector_pipeline(
    qvec, *, limit, candidates=None, vector_index=VECTOR_INDEX_NAME, status_in=DECIDED_STATUSES
) -> list[dict]:
    return [
        {
            "$vectorSearch": {
                "index": vector_index,
                "path": "embedding",
                "queryVector": qvec,
                "numCandidates": candidates or max(50, limit * 10),
                "limit": limit,
                "filter": {"status": {"$in": list(status_in)}},
            }
        },
        _project_stage(),
    ]


def build_lexical_pipeline(query, *, limit, search_index=SEARCH_INDEX_NAME) -> list[dict]:
    return [
        {"$search": {"index": search_index, "text": {"query": query, "path": LEXICAL_PATHS}}},
        {"$limit": limit},
        _project_stage(),
    ]


def build_rank_fusion_pipeline(
    qvec,
    query,
    *,
    k,
    vector_index=VECTOR_INDEX_NAME,
    search_index=SEARCH_INDEX_NAME,
    status_in=DECIDED_STATUSES,
) -> list[dict]:
    # $rankFusion (MongoDB 8.0+) runs both input pipelines and fuses them by reciprocal rank —
    # one aggregation, server-side. Each input pipeline gets weight 1 (uniform). To bias toward
    # semantic vs. lexical, add "combination": {"weights": {"vector": w, "lexical": w}}.
    candidates = max(50, k * 10)
    per_branch = max(k * 4, 20)
    return [
        {
            "$rankFusion": {
                "input": {
                    "pipelines": {
                        "vector": [
                            {
                                "$vectorSearch": {
                                    "index": vector_index,
                                    "path": "embedding",
                                    "queryVector": qvec,
                                    "numCandidates": candidates,
                                    "limit": per_branch,
                                    "filter": {"status": {"$in": list(status_in)}},
                                }
                            }
                        ],
                        "lexical": [
                            {
                                "$search": {
                                    "index": search_index,
                                    "text": {"query": query, "path": LEXICAL_PATHS},
                                }
                            },
                            {"$limit": per_branch},
                        ],
                    }
                },
            }
        },
        {"$limit": k},
        _project_stage(with_score=True),
    ]


def build_graph_pipeline(
    account_id: str, *, max_depth: int = 4, collection: str = "transactions"
) -> list[dict]:
    return [
        {"$match": {"sender.account_number": account_id}},
        {
            "$graphLookup": {
                "from": collection,
                "startWith": "$recipient.account_number",
                "connectFromField": "recipient.account_number",
                "connectToField": "sender.account_number",
                "as": "chain",
                "maxDepth": max_depth,
                "depthField": "depth",
            }
        },
    ]


def summarize_ring(graph_doc: dict, *, seed_account: str) -> dict:
    # Turn a $graphLookup chain into fraud-ring signals: circular flow back to the seed account,
    # layering (many small transfers), and overall network size.
    chain: list[dict] = list(graph_doc.get("chain", []))
    accounts: set[str] = set()
    small_transfers = 0
    circular_flow = False
    for edge in chain:
        sender = (edge.get("sender") or {}).get("account_number")
        recipient = (edge.get("recipient") or {}).get("account_number")
        accounts.update(a for a in (sender, recipient) if a)
        if recipient == seed_account:
            circular_flow = True
        if float(edge.get("amount", 0) or 0) < 1000:
            small_transfers += 1
    network_size = len(chain)
    layering = small_transfers >= 5
    return {
        "network_size": network_size,
        "unique_accounts": len(accounts),
        "circular_flow": circular_flow,
        "layering": layering,
        "suspicious_patterns": circular_flow or layering or network_size >= 3,
    }

docs = prepare_seed(load_seed(), now=datetime.now(UTC))
count = seed_collection(coll, docs)
print(f"seeded {count} transactions")

# Create the vector + Atlas Search indexes if absent, then block until both are queryable AND
# actually reflect the freshly-seeded docs (Atlas Search is eventually consistent).
ensure_indexes(coll, dim=EMBED_DIM)

check = preflight(coll)
for issue in check["issues"]:
    print("PREFLIGHT:", issue)
assert check["ok"], "Fix the preflight issues above before continuing."

# This cookbook uses the native `$rankFusion` stage for hybrid search, which needs MongoDB 8.0+.
# Every current Atlas cluster — including the free M0 tier — runs 8.0 or later, so this holds by
# default; the check just fails loudly if you point MONGO_URI at an older self-hosted server.
version = server_version(coll)
assert supports_rank_fusion(version), (
    f"MongoDB {version} predates $rankFusion (needs 8.0+). Use an Atlas cluster (any tier) or a "
    f"self-hosted MongoDB 8.0+."
)
print(f"server={version}  hybrid search=$rankFusion (native)")

seeded 20 transactions
server=8.0.27  hybrid search=$rankFusion (native)

语义召回：找到与查询向量最近的embedding的文档。此示例

重新使用现有文档的嵌入作为查询（一个寻找已决先例的待决案件），因此不需要嵌入API调用。构建器通过status字段过滤到已决案件。 ### 模式2：全文搜索（$search，BM25）文本字段中的关键词和短语匹配——这是捕捉嵌入模糊的精确名称、ID和代码的方法。

query_doc = coll.find_one({"transaction_id": "txn-review-struct"})  # a pending case
hits = list(coll.aggregate(build_vector_pipeline(query_doc["embedding"], limit=3)))
print("nearest decided precedents:")
for h in hits:
    print(f"  {h['transaction_id']}: {h['text'][:72]}")

nearest decided precedents:
  txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third
  txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account
  txn-struct-03: Transfer of 4999 USD deliberately under 5000 to avoid reporting. Pattern
```### 模式3：混合（互逆排名融合）

Keyword and phrase matching over text fields — this is what catches the exact names, IDs, and
codes that embeddings blur.

### 关键词和短语匹配在文本字段上——这正是捕捉到嵌入模糊的精确名称、ID和代码的方法。```python
hits = list(
    coll.aggregate(
        build_lexical_pipeline("cash deposit just under the reporting threshold", limit=3)
    )
)
print("full-text matches:")
for h in hits:
    print(f"  {h['transaction_id']}: {h['text'][:72]}")

full-text matches:
  txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third
  txn-review-struct: Cash deposit of 4950 USD, just under the 5000 reporting threshold, follo
  txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account

Pattern 3: Hybrid (reciprocal rank fusion)

将向量和全文排名融合成一个结果集。$rankFusion（MongoDB 8.0+）在服务器端运行同时处理输入管道并在互相关排名的基础上合并它们——无需客户端合并， no second round-trip.

qvec, query = query_doc["embedding"], "structuring: cash deposit just under the threshold"
hits = list(coll.aggregate(build_rank_fusion_pipeline(qvec, query, k=3)))
print("hybrid via $rankFusion (server-side):")
for h in hits:
    print(f"  {h['transaction_id']}: {h['text'][:72]}")

hybrid via $rankFusion (server-side):
  txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third
  txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account
  txn-struct-03: Transfer of 4999 USD deliberately under 5000 to avoid reporting. Pattern

模式 4：图遍历 (`$graphLookup`)

跟踪 sender.account_number -> recipient.account_number 链接以展示一个网络（这里，是一个环形资金流动环）。这是一个关系信号，故意与以下内容分开 similarity ranking above.

graph_doc = next(
    iter(coll.aggregate(build_graph_pipeline("ACC-RING-A", collection=coll.name))), {"chain": []}
)
ring = summarize_ring(graph_doc, seed_account="ACC-RING-A")
print(
    f"graph traversal from ACC-RING-A: network_size={ring['network_size']} "
    f"circular_flow={ring['circular_flow']} suspicious={ring['suspicious_patterns']}"
)

graph traversal from ACC-RING-A: network_size=4 circular_flow=True suspicious=True

根据您的领域调整这些设置。 每个构建器都是一个通用形状：将您的文本嵌入到任何模型中，并将 build_vector_pipeline 指向您的字段；设置全文的搜索路径；调整 $rankFusion 输入权重以偏重语义与词汇；更改图 connectFromField/connectToField 以匹配您的关系（引用、组织结构图、供应链）。所有四个都在上面的单元格中定义——直接将它们应用到您自己的项目中。

3. 端到端代理：人工参与欺诈审查

现在一切协同工作。一个托管代理审查标记的交易，每个建议都基于第 2 节中的 MongoDB 信号（作为 自定义工具 在路径 A 的往返中公开），并在有风险的案例中暂停以供人工处理。决策和只读审计跟踪被写回到同一个集群。代理循环运行在Anthropic上；数据路径在您的笔记本中运行。MongoDB Atlas提供检索信号并存储决策和审计跟踪。

代理循环运行在Anthropic上；数据路径在您的笔记本中运行。因此，

MONGO_URI 和嵌入密钥永远不会进入代理或其沙盒。人工等待是持久的——会话在服务器端空闲，通过 requires_action 直到您响应——而下面的简单流式循环用于开发（通过 webhook 从持久的后端驱动它）。 session sits idle server-side via requires_action until you respond — while the simple streaming loop below is for development (drive it from a durable backend with the webhook pattern in production).

AP2强制要求：验证后再推理

此代理审查由代理发起的付款，因此在任何行为分析之前，它都会验证一个 AP2（代理付款协议）强制要求——这是加密签名的凭证，证明用户授权了付款。AP2是一个开放协议用于代理驱动的商业。其强制要求（一个结账强制要求和一个付款强制要求）是可篡改的，签名的数字对象，形成一个可审计的链。您不需要学习该协议就可以遵循这个食谱：调用verify_mandates工具并采取其裁决 （有效、满足约束和双重支出）。签名和验证的详细信息——ES256 JWTs—— 保留在ap2_mandates.py中，签名的强制要求和收据存储在MongoDB中，与交易一起。下面的单元格扮演“可信表面”的角色（生成密钥对，将签名的强制要求附加到待处理案件）。私钥保留在本地Python状态中——永远不会是代理、工具结果或

自定义工具和主机端处理程序

每个工具都是一个type: "custom"工具。当代理调用一个时，会话暂停，这个笔记本使用pymongo运行处理程序，并将JSON结果发送回——代理永远不会 the database.

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import ec

ts_private_key = ec.generate_private_key(ec.SECP256R1())  # Trusted Surface key, fresh each run
AGENT_PK = (
    ts_private_key.public_key()
    .public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
    .hex()
)
ts_public_key = ts_private_key.public_key()

PENDING = [
    "txn-review-clean",
    "txn-review-fraud",
    "txn-review-struct",  # seeded human-override case under AUTO_APPROVE
    "txn-review-high",
    "txn-review-ring",
]
mandates = attach_mandates(coll, PENDING, agent_pk=AGENT_PK, ts_private_key=ts_private_key)
print(f"attached AP2 mandates to {len(mandates)} transactions  agent_pk={AGENT_PK[:16]}…")

attached AP2 mandates to 5 transactions  agent_pk=04e1e32b7414d6f7…
```### 创建代理、环境和会话

每个工具都是一个 `type: "custom"` 工具。当代理调用其中一个时，会话暂停，**此笔记本**使用 `pymongo` 运行处理器，并发送JSON结果回传——代理永远不会接触数据库或连接字符串。处理器实现定义在下一单元格中：它们封装了第2节的构建器以及AP2验证（`verify_mandates`）。`HANDLERS` 只映射五个数据工具；`escalate` 故意省略，因为它被门循环路由到人工解析器而不是处理器（见下一节）。

```python
def handle_data(data):
    # 处理数据
    pass

{
    "result": "success"
}

SELECT * FROM users WHERE id = 1;
``````python
# The host-side tool implementations — the pymongo work behind each custom tool. These run in
# THIS process (never the sandbox), reusing the Section 2 builders. The dispatch wiring + the
# AP2/record-decision wrappers are in the next cell; build_decision_doc / build_audit_event
# (imported above) shape the persisted documents and are shared with the AP2 module.


def _jsonable(value):
    """Recursively coerce Mongo/BSON values (ObjectId, datetime, ...) into JSON-safe types."""
    if isinstance(value, dict):
        return {k: _jsonable(v) for k, v in value.items()}
    if isinstance(value, (list, tuple)):
        return [_jsonable(v) for v in value]
    if isinstance(value, datetime):
        return value.isoformat()
    if isinstance(value, (str, int, float, bool)) or value is None:
        return value
    return str(value)


def _pick_fields(candidate):
    keep = ("transaction_id", "text", "amount", "sender", "recipient", "decision", "score")
    return _jsonable({k: candidate[k] for k in keep if k in candidate})


def rerank_pool_size(k: int, fanout: int, *, enable_rerank: bool) -> int:
    return k * fanout if enable_rerank else k


def merge_rerank(candidates, rerank_results, *, top_k) -> list[dict]:
    def _score(r):
        s = getattr(r, "relevance_score", None)
        return (
            (s if s is not None else r.get("relevance_score", float("-inf")))
            if isinstance(r, dict)
            else (s if s is not None else float("-inf"))
        )

    seen: set[str] = set()
    out: list[dict] = []
    for result in sorted(rerank_results, key=_score, reverse=True):
        index = getattr(result, "index", None)
        if index is None and isinstance(result, dict):
            index = result.get("index")
        score = getattr(result, "relevance_score", None)
        if score is None and isinstance(result, dict):
            score = result.get("relevance_score")
        if index is None or index < 0 or index >= len(candidates):
            continue
        candidate = candidates[index]
        cid = str(candidate.get("transaction_id", index))
        if cid in seen:
            continue
        seen.add(cid)
        out.append({**candidate, "score": score})
        if len(out) >= top_k:
            break
    return out


def tool_get_transaction(coll, transaction_id) -> dict:
    doc = coll.find_one({"transaction_id": transaction_id})
    if not doc:
        return {"error": "not_found", "transaction_id": transaction_id}
    return _jsonable({k: v for k, v in doc.items() if k not in ("embedding", "_id")})


def tool_hybrid_search_similar_frauds(
    coll,
    transaction_id,
    k,
    *,
    enable_rerank=False,
    fanout=5,
    reranker=None,
    vector_index=VECTOR_INDEX_NAME,
    search_index=SEARCH_INDEX_NAME,
    status_in=DECIDED_STATUSES,
) -> dict:
    txn = coll.find_one({"transaction_id": transaction_id})
    if not txn:
        return {"error": "not_found", "transaction_id": transaction_id, "similar": []}
    qvec = txn["embedding"]
    query = txn.get("text", "")
    pool_k = rerank_pool_size(k, fanout, enable_rerank=enable_rerank)
    candidates = list(
        coll.aggregate(
            build_rank_fusion_pipeline(
                qvec,
                query,
                k=pool_k,
                vector_index=vector_index,
                search_index=search_index,
                status_in=status_in,
            )
        )
    )
    candidates = [c for c in candidates if str(c.get("transaction_id")) != str(transaction_id)]
    if enable_rerank and reranker is not None and candidates:
        results = reranker(query, [c.get("text", "") for c in candidates], top_k=k)
        candidates = merge_rerank(candidates, results, top_k=k)
    else:
        candidates = candidates[:k]
    return {"similar": [_pick_fields(c) for c in candidates]}


def tool_detect_fraud_ring(coll, account_id, *, max_depth=4) -> dict:
    pipeline = build_graph_pipeline(account_id, max_depth=max_depth, collection=coll.name)
    docs = list(coll.aggregate(pipeline))
    return _jsonable(summarize_ring(docs[0] if docs else {"chain": []}, seed_account=account_id))


def tool_record_decision(
    db,
    transaction_id,
    decision,
    *,
    confidence,
    risk_factors,
    reasoning,
    reviewed_by,
    escalated=False,
    recommended_decision=None,
) -> dict:
    decision_doc = build_decision_doc(
        transaction_id,
        decision,
        confidence=confidence,
        risk_factors=risk_factors,
        reasoning=reasoning,
        reviewed_by=reviewed_by,
    )
    db["transaction_decisions"].insert_one(decision_doc)
    if escalated:
        audit = build_audit_event(
            "escalated_to_human",
            transaction_id,
            decision_id=decision_doc["decision_id"],
            severity="warning",
            event_data={"human_decision": decision, "recommended_decision": recommended_decision},
        )
    else:
        audit = build_audit_event(
            "decision_stored", transaction_id, decision_id=decision_doc["decision_id"]
        )
    db["audit_events"].insert_one(audit)
    # Advance the lifecycle status to past tense (approve -> approved) so a decided case matches
    # the seed's vocabulary and DECIDED_STATUSES, making it eligible as precedent in later reviews.
    status = {"approve": "approved", "reject": "rejected"}.get(decision, decision)
    db["transactions"].update_one({"transaction_id": transaction_id}, {"$set": {"status": status}})
    return {"recorded": True, "decision_id": decision_doc["decision_id"]}

TOOLS = [
    {
        "type": "custom",
        "name": "verify_mandates",
        "description": "Validate the AP2 Checkout and Payment Mandate JWTs (signature, constraints, "
        "double-spend). Run this FIRST. If valid=false, constraints_satisfied=false, or "
        "double_spend_detected=true, reject immediately.",
        "input_schema": {
            "type": "object",
            "properties": {"transaction_id": {"type": "string"}},
            "required": ["transaction_id"],
        },
    },
    {
        "type": "custom",
        "name": "get_transaction",
        "description": "Fetch the full transaction record under review.",
        "input_schema": {
            "type": "object",
            "properties": {"transaction_id": {"type": "string"}},
            "required": ["transaction_id"],
        },
    },
    {
        "type": "custom",
        "name": "hybrid_search_similar_frauds",
        "description": "Retrieve the most similar prior (already-decided) cases as precedent, using "
        "hybrid vector + full-text search.",
        "input_schema": {
            "type": "object",
            "properties": {
                "transaction_id": {"type": "string"},
                "k": {"type": "integer", "description": "how many precedents (default 5)"},
            },
            "required": ["transaction_id"],
        },
    },
    {
        "type": "custom",
        "name": "detect_fraud_ring",
        "description": "Trace the account's sender->recipient chain for circular-flow / mule / layering patterns.",
        "input_schema": {
            "type": "object",
            "properties": {"account_id": {"type": "string"}},
            "required": ["account_id"],
        },
    },
    {
        "type": "custom",
        "name": "record_decision",
        "description": "Persist the final approve/reject decision with reasoning and an audit event.",
        "input_schema": {
            "type": "object",
            "properties": {
                "transaction_id": {"type": "string"},
                "decision": {"type": "string", "enum": ["approve", "reject"]},
                "confidence": {"type": "number"},
                "risk_factors": {"type": "array", "items": {"type": "string"}},
                "reasoning": {"type": "string"},
                "escalated": {"type": "boolean"},
                "recommended_decision": {"type": "string", "enum": ["approve", "reject"]},
            },
            "required": ["transaction_id", "decision", "reasoning"],
        },
    },
    {
        "type": "custom",
        "name": "escalate",
        "description": "Send a risky case to a human reviewer for the final decision. Use for "
        "medium-confidence, high-value, structuring, or fraud-ring cases.",
        "input_schema": {
            "type": "object",
            "properties": {
                "transaction_id": {"type": "string"},
                "recommended_decision": {"type": "string", "enum": ["approve", "reject"]},
                "confidence": {"type": "number"},
                "reason": {"type": "string"},
            },
            "required": ["transaction_id", "recommended_decision", "reason"],
        },
    },
]

# Host-side handlers — map each data tool to its handler function.
reranker = (
    (lambda q, ds, top_k: rerank(q, ds, client=ai_client, top_k=top_k))
    if (ENABLE_RERANK and ai_client)
    else None
)


def _verify_mandates(inp):
    result = tool_verify_mandates(db, inp["transaction_id"], ts_public_key)
    ok = result["valid"] and result["constraints_satisfied"] and not result["double_spend_detected"]
    print(f"\n  [verify_mandates] {inp['transaction_id']}: {'pass' if ok else 'FAIL'}")
    return result


def _hybrid(inp):
    return tool_hybrid_search_similar_frauds(
        coll,
        inp["transaction_id"],
        inp.get("k", 5),
        enable_rerank=ENABLE_RERANK,
        reranker=reranker,
    )


def _record(inp):
    result = tool_record_decision(
        db,
        inp["transaction_id"],
        inp["decision"],
        confidence=inp.get("confidence", 0),
        risk_factors=inp.get("risk_factors", []),
        reasoning=inp.get("reasoning", ""),
        reviewed_by="human" if inp.get("escalated") else "agent",
        escalated=inp.get("escalated", False),
        recommended_decision=inp.get("recommended_decision"),
    )
    if (
        inp["decision"] == "approve"
    ):  # store the AP2 receipt that later powers double-spend detection
        txn_doc = coll.find_one(
            {"transaction_id": inp["transaction_id"]},
            {"checkout_mandate_jwt": 1, "mandate_id": 1, "agent_pk": 1},
        )
        if txn_doc and txn_doc.get("mandate_id"):
            checkout_hash = hashlib.sha256(txn_doc["checkout_mandate_jwt"].encode()).hexdigest()
            store_mandate_receipt(
                db, txn_doc["mandate_id"], txn_doc["agent_pk"], checkout_hash, "approve"
            )
    return result


HANDLERS = {
    "verify_mandates": _verify_mandates,
    "get_transaction": lambda inp: tool_get_transaction(coll, inp["transaction_id"]),
    "hybrid_search_similar_frauds": _hybrid,
    "detect_fraud_ring": lambda inp: tool_detect_fraud_ring(coll, inp["account_id"]),
    "record_decision": _record,
}

Create the agent, environment, and session

model、system和tools都存在于agent上。会话引用它并配置沙盒。网络是有限的：agent只能通过主机端的往返访问MongoDB。使用人类门控运行审查

SYSTEM = """You are a financial fraud reviewer. You will be given transaction IDs to review.

For EACH transaction, in order:
1. Call get_transaction to read it.
2. Call verify_mandates to validate the AP2 mandate chain (signature, constraints, double-spend).
   HARD GATE: if verify_mandates returns valid=false, constraints_satisfied=false, OR
   double_spend_detected=true, call record_decision immediately with decision="reject" and
   skip the remaining steps for that transaction.
3. Call hybrid_search_similar_frauds to retrieve similar decided precedents.
4. Call detect_fraud_ring on the sender's account_number to check for ring/mule patterns.
5. Weigh the precedents, the ring signal, the amount, and your confidence. Then make EXACTLY
   ONE terminal call for that transaction:
   - You MUST call escalate (do NOT call record_decision yourself) whenever ANY of these holds:
       * a structuring amount ($4,900-$4,999),
       * a high-value amount (>= $50,000) that you would otherwise approve,
       * detect_fraud_ring reports suspicious_patterns, or
       * your confidence is medium (~75-85).
     Give your recommended_decision and a short reason.
   - Otherwise (a clear-cut case matching none of the above), call record_decision (approve/reject).

When you escalate, you will receive the human's decision. Then call record_decision with that
decision, escalated=true, and recommended_decision set to what you had recommended.

Be concise. Move to the next transaction after recording a decision."""

agent = client.beta.agents.create(
    name="MongoDB Atlas fraud reviewer", model=MODEL, system=SYSTEM, tools=TOOLS
)
environment = client.beta.environments.create(
    name="fraud-review-env", config={"type": "cloud", "networking": {"type": "limited"}}
)
session = client.beta.sessions.create(
    agent={"type": "agent", "id": agent.id, "version": agent.version},
    environment_id=environment.id,
    title="Fraud review",
)
print("session:", session.id)
print(f"Watch in Console: https://platform.claude.com/workspaces/default/sessions/{session.id}")

session: sesn_01LbAUC1V36CQS3Sk26eixTc
Watch in Console: https://platform.claude.com/workspaces/default/sessions/sesn_01LbAUC1V36CQS3Sk26eixTc

打开事件流，发送待处理事务队列，并使用run_gate_loop驱动门控循环：数据工具调用会自动从HANDLERS服务，而一个escalate调用会暂停以供人类处理。当设置AUTO_APPROVE（例如，在CI中）时，门控会确定性地解决，并覆盖了agent在种子结构案例上的设置，因此可以看到不同的人类裁决。否则，您将直接收到提示。

审查决策和审计跟踪从MongoDB读取决策和审计跟踪。决策和审计集合是只读的，因此读取范围仅限于本次运行。关闭断言使这成为一个自我检查：如果重新运行没有重现预期的通道混合，则会大声失败。人类覆盖计数反映了门控是如何解决的：在AUTO_APPROVE中为1（txn-review-struct上的种子覆盖显示），而在交互式运行中，每当您同意agent的推荐并每次升级时，它都是0——两者都是预期的。 4. MongoDB Atlas作为记录系统和审计骨干回顾运行状态所在的位置：agent所做的一切都在它检索的同一集群中是持久的。这是写入端的单一引擎回报——transactions是操作记录（每个案例推进pending→approved或rejected），transaction_decisions持有不可变的裁决，audit_events是一个只读的

OVERRIDE_IDS = {"txn-review-struct"}


def resolve_human_decision(recommendation, *, auto_approve, override_ids, txn_id) -> str:
    """Stand in for a human reviewer. In CI (auto_approve) it concurs with the agent, except on
    the seeded override_ids where it flips the verdict so a differing human decision is visible."""
    opposite = "reject" if recommendation == "approve" else "approve"
    if auto_approve and txn_id in set(override_ids):
        return opposite
    return recommendation


def resolver(tool_input):
    txn_id = tool_input.get("transaction_id", "")
    recommended = tool_input.get("recommended_decision", "reject")
    if AUTO_APPROVE:
        decision = resolve_human_decision(
            recommended, auto_approve=True, override_ids=OVERRIDE_IDS, txn_id=txn_id
        )
        flag = "  <-- HUMAN OVERRIDE" if decision != recommended else ""
        print(f"  [human/auto] {txn_id}: agent recommended {recommended} -> human {decision}{flag}")
        return decision
    print(
        f"\n  ESCALATED {txn_id} (agent recommends {recommended}): {tool_input.get('reason', '')}"
    )
    return (
        "approve"
        if input("  Your decision [approve/reject]: ").strip().lower().startswith("a")
        else "reject"
    )


def send_result(custom_tool_use_id, result):
    client.beta.sessions.events.send(
        session_id=session.id,
        events=[
            {
                "type": "user.custom_tool_result",
                "custom_tool_use_id": custom_tool_use_id,
                "content": [{"type": "text", "text": json.dumps(result)}],
            }
        ],
    )


def run_gate_loop(stream, send, *, handlers, resolver) -> dict:
    """Drive the session to completion, servicing custom-tool calls as they arrive.

    This is the whole Path A round-trip for a real workload. Each `agent.custom_tool_use`
    event names a tool and its input; the session then goes idle with
    `stop_reason == "requires_action"` until we answer. We look up the tool in `handlers`
    (which run `pymongo` host-side — the credential never leaves this process), and route the
    special `escalate` tool to the human `resolver` instead. We post each result back with
    `send`, and the session resumes. Break on a terminal idle (`end_turn`) or termination.
    """
    pending: dict[str, Any] = {}
    responded: set[str] = set()
    serviced: list[tuple[str, dict]] = []
    for ev in stream:
        if ev.type == "agent.custom_tool_use":
            pending[ev.id] = ev
        elif ev.type == "session.status_idle":
            stop = getattr(ev, "stop_reason", None)
            if stop is None:
                continue
            if stop.type != "requires_action":
                break  # end_turn / retries_exhausted — the run is done
            for event_id in stop.event_ids or []:
                if event_id in responded:
                    continue
                call = pending.get(event_id)
                if call is None:
                    continue
                if call.name == "escalate":
                    result = {"human_decision": resolver(call.input)}
                else:
                    handler = handlers.get(call.name)
                    result = (
                        handler(call.input) if handler else {"error": f"unknown tool {call.name}"}
                    )
                responded.add(event_id)
                send(event_id, result)
                serviced.append((call.name, result))
        elif ev.type == "session.status_terminated":
            break
    return {"serviced": serviced, "responded": sorted(responded)}


kickoff = {
    "type": "user.message",
    "content": [
        {"type": "text", "text": "Review these flagged transactions: " + ", ".join(PENDING)}
    ],
}
run_start = datetime.now(UTC)

with client.beta.sessions.events.stream(session_id=session.id) as stream:
    client.beta.sessions.events.send(session_id=session.id, events=[kickoff])
    gate_result = run_gate_loop(stream, send_result, handlers=HANDLERS, resolver=resolver)

wait_for_idle_status(client, session.id)
print(f"\nserviced {len(gate_result['serviced'])} tool calls")


  [verify_mandates] txn-review-clean: pass

  [verify_mandates] txn-review-fraud: pass

  [verify_mandates] txn-review-struct: pass

  [verify_mandates] txn-review-high: pass

  [verify_mandates] txn-review-ring: pass

  ESCALATED txn-review-struct (agent recommends reject): Structuring indicator: cash deposit of $4,950 just under $5,000 CTR threshold, matching precedent (txn-struct-01, txn-struct-02) showing pattern of deliberate sub-threshold deposits from same account. Amount in structuring range per policy. Requires human review.

  Your decision [approve/reject]:  reject


  ESCALATED txn-review-high (agent recommends approve): High-value wire of $75,000 from Northwind Foods to Pacific Logistics for bulk shipment. Legitimate business transaction matching precedent txn-high-01 ($120,000 supplier wire, approved). Mandate valid, no fraud signals. Escalation required per policy for high-value transactions ≥ $50,000.

  Your decision [approve/reject]:  approve


  ESCALATED txn-review-ring (agent recommends reject): Money-mule ring detected. Sender ACC-RING-A (Quartz Trading) is part of a 4-account network with circular flows. Precedent txn-ring-01 (same sender/recipient, $920, rejected) shows identical pattern. Fraud ring analysis confirms suspicious_patterns, circular_flow. Requires human review for definitive ring-case determination.

  Your decision [approve/reject]:  reject


serviced 28 tool calls
```### 审查决策和审计跟踪

从MongoDB读取决策和审计跟踪。决策和审计集合是**只读**的，因此读取范围仅限于本次运行。关闭断言使这成为一个**自我检查**：如果重新运行没有重现预期的通道混合，则会大声失败。

人工干预计数反映了门是如何解决的：使用`AUTO_APPROVE`时为1（`txn-review-struct`上的种子干预显示），而在交互式运行中，每当您同意代理的推荐并解决每个升级时，它都是0——两者都是预期的。

- **append-only**：只追加
- **self-check**：自我检查
- **AUTO_APPROVE**：自动批准```python
from collections import Counter

decisions = list(db["transaction_decisions"].find({"created_at": {"$gte": run_start}}, {"_id": 0}))
audits = list(db["audit_events"].find({"timestamp": {"$gte": run_start}}, {"_id": 0}))

print("Decisions by lane:")
lanes = Counter()
for d in decisions:
    txn = coll.find_one({"transaction_id": d["transaction_id"]})
    lanes[f"{(txn['lane'] if txn else '?')} -> {d['decision']}"] += 1
for k, v in sorted(lanes.items()):
    print(f"  {k}: {v}")

overrides = [
    a
    for a in audits
    if a["event_type"] == "escalated_to_human"
    and a["event_data"].get("human_decision") != a["event_data"].get("recommended_decision")
]
print(
    f"\nescalated_to_human events: {sum(1 for a in audits if a['event_type'] == 'escalated_to_human')}"
)
print(f"human overrides (verdict != agent recommendation): {len(overrides)}")
for a in overrides:
    ed = a["event_data"]
    print(
        f"  {a['transaction_id']}: agent {ed['recommended_decision']} -> human {ed['human_decision']}"
    )

# Self-check — fail loudly if this run didn't produce the expected result.
decided_lanes = {coll.find_one({"transaction_id": d["transaction_id"]})["lane"] for d in decisions}
assert decided_lanes == {"clean_approve", "clear_reject", "high_value", "ring", "structuring"}, (
    f"expected all 5 lanes decided, saw {sorted(decided_lanes)}"
)
assert len(decisions) == len(PENDING), f"expected {len(PENDING)} decisions, got {len(decisions)}"
assert sum(1 for a in audits if a["event_type"] == "escalated_to_human") >= 1, (
    "expected >=1 escalation"
)
if AUTO_APPROVE:
    assert overrides, "AUTO_APPROVE run should show the seeded override on txn-review-struct"
print("\nself-check OK")

Decisions by lane:
  clean_approve -> approve: 1
  clear_reject -> reject: 1
  high_value -> approve: 1
  ring -> reject: 1
  structuring -> reject: 1

escalated_to_human events: 3
human overrides (verdict != agent recommendation): 0

self-check OK
```## 4. MongoDB Atlas 作为记录和审计骨干系统

退后一步，看看运行状态存储在哪里：代理所做的一切都在它检索的同一个集群中是持久的。这是写入端的单一引擎回报——`transactions` 是操作记录（每个案例推进 `pending` → `approved` 或 `rejected`），`transaction_decisions` 存储不可变的裁决，`audit_events` 是一个只追加跟踪，以及 `mandate_receipts` 权限用于双重消费检测。一个案例，从头到尾：

::: callout-注意
:::
请注意，以下代码示例仅用于演示目的，实际应用中可能需要根据具体情况进行调整。
:::

```python
# 示例代码
def process_transaction(transaction_id):
    # 处理事务逻辑
    pass

{
    "transaction_id": "123456",
    "status": "pending"
}

SELECT * FROM transactions WHERE status = 'pending';
``````python
case = "txn-review-struct"
txn = coll.find_one({"transaction_id": case}, {"_id": 0, "embedding": 0})
print(f"operational record: {case}  lane={txn['lane']}  status={txn['status']}")

decision = db["transaction_decisions"].find_one(
    {"transaction_id": case, "created_at": {"$gte": run_start}}, {"_id": 0}
)
print(
    f"decision record:    {decision['decision']}  reviewed_by={decision['reviewed_by']}  confidence={decision['confidence_score']}"
)

print("audit trail (append-only):")
for a in (
    db["audit_events"]
    .find({"transaction_id": case, "timestamp": {"$gte": run_start}}, {"_id": 0})
    .sort("timestamp", 1)
):
    extra = ""
    if a["event_type"] == "escalated_to_human":
        ed = a["event_data"]
        extra = f"  (agent recommended {ed['recommended_decision']}, human decided {ed['human_decision']})"
    print(f"  {a['timestamp']:%H:%M:%S}  {a['event_type']}  severity={a['severity']}{extra}")

txn_mandate = coll.find_one({"transaction_id": case}, {"mandate_id": 1, "agent_pk": 1})
if txn_mandate and txn_mandate.get("mandate_id"):
    receipt = db["mandate_receipts"].find_one({"mandate_id": txn_mandate["mandate_id"]}, {"_id": 0})
    if receipt:
        print(
            f"mandate receipt:    mandate_id={receipt['mandate_id'][:12]}…  decision={receipt['decision']}"
        )

operational record: txn-review-struct  lane=structuring  status=rejected
decision record:    reject  reviewed_by=human  confidence=0.9
audit trail (append-only):
  13:49:22  escalated_to_human  severity=warning  (agent recommended reject, human decided reject)

生产环境中的Webhooks（指针，此处未运行）： 上面的流循环非常适合开发，但无法在重启后存活。对于生产环境，为 session.status_idled 注册一个Webhook：你的处理程序将案例排队给审阅者，稍后通过POST发送 user.custom_tool_result。数据工具处理程序是相同的；只有触发器不同。查看 CMA_operate_in_production.ipynb。您将MongoDB Atlas连接到Claude托管代理，并从连接到系统： - 连接 — 三种凭证安全的数据路径（主机端自定义工具、自托管沙盒、自托管MCP），MongoDB密钥始终在您的边界一侧。 - 检索 — 向量、全文、混合RRF和图遍历作为可提升的管道构建器，在第二章中定义。

client.beta.sessions.archive(session.id)
client.beta.environments.archive(environment.id)
# Agents are reusable across runs; archiving is optional and permanent.
mongo.close()
print("archived session + environment; closed MongoDB connection")

archived session + environment; closed MongoDB connection

5. Recap

门控 — CMA原生 requires_action 人类暂停背后的风险决策。 record in one build:
持久化 — 决策、仅追加的审计跟踪和AP2收据在同一集群中。集成界面很小：自定义工具背后的几行 pymongo 代码。交换集合和工具，相同的形状让任何代理在MongoDB中都能运行。检索构建器、工具处理程序和门控循环都直接在这里的笔记本中；mongodb_on_cma/ [9] 包只包含设置模板。
Retrieve — vector, full-text, hybrid RRF, and graph traversal as liftable pipeline builders, defined inline in Section 2.
Gate — risky decisions behind CMA’s native requires_action human pause.
Persist — decisions, an append-only audit trail, and AP2 receipts in the same cluster the agent retrieves from.

The integration surface was small: a few lines of pymongo behind a custom tool. Swap the collection and the tools, and the same shape grounds any agent in MongoDB. The retrieval builders, tool handlers, and gate loop are all right here in the notebook; the mongodb_on_cma/ package holds only the setup boilerplate.

Tip

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

基于 MongoDB Atlas 和 Claude 管理代理的欺诈审查代理

本指南涵盖的内容

设置

Quiet the SDK’s one-shot notice that ANTHROPIC_API_KEY shadows profile/federation

auto-discovery — expected when an API key is set; harmless under ant / WIF auth.

Other options (brief)

2. Retrieve: four patterns, one engine

Pattern 3: Hybrid (reciprocal rank fusion)

模式 4：图遍历 ($graphLookup)

3. 端到端代理：人工参与欺诈审查

AP2强制要求：验证后再推理

自定义工具和主机端处理程序

Create the agent, environment, and session

5. Recap

auto-discovery — expected when an API key is set; harmless under `ant` / WIF auth.

模式 4：图遍历 (`$graphLookup`)