使用 MongoDB Atlas 和 Claude 托管代理构建欺诈审查代理

本教程面向有编程基础的开发者，深入讲解如何将 MongoDB Atlas 集成到 Claude 托管代理（CMA）中，作为检索引擎、图存储和记录系统。通过主机端自定义工具实现凭据安全的数据路径，利用 MongoDB 的向量搜索、全文搜索、混合搜索（RRF）和图遍历四种检索模式，构建一个带人工审批门控的欺诈审查代理。教程涵盖环境准备、核心原理、架构设计、代码分析、常见错误排查及生产优化，帮助开发者掌握 AI 工程实践中的关键集成技术。

技术背景

在构建 AI 代理时，开发者常面临检索、存储和审计的碎片化问题：需要组合向量数据库、全文搜索引擎、图数据库和操作数据库，每个组件都带来额外的集成复杂度、凭据管理和数据一致性挑战。MongoDB Atlas 通过统一引擎解决了这一问题，支持向量、全文、混合和图检索，同时作为操作数据存储。本教程旨在展示如何利用 MongoDB 的这一能力，结合 Claude 托管代理的托管循环和沙箱，构建一个安全、可审计的欺诈审查代理，并强调凭据安全边界和人在环路（HITL）的重要性。

核心原理

核心原理是数据路径与代理循环的分离：代理循环在 Anthropic 托管环境中运行，而 MongoDB 凭据和数据访问完全保留在开发者主机端。通过自定义工具（custom tools）实现主机端查询，代理调用工具时，会话暂停（requires_action），主机端执行 pymongo 查询并返回结果，凭据永不进入代理上下文。MongoDB 提供四种检索模式：$vectorSearch（向量相似度）、$search（全文 BM25）、$rankFusion（混合倒数排名融合）和 $graphLookup（图遍历），均通过聚合管道实现。代理通过工具调用这些管道，实现语义搜索、关键词匹配、混合排序和关系发现。

架构分析

系统架构分为三层：代理层（Claude 托管代理）、主机端数据路径层（自定义工具处理器）和 MongoDB Atlas 层。代理层包含代理定义（模型、系统提示、工具列表）和会话管理，通过 Anthropic SDK 的 beta 接口驱动。主机端数据路径层是核心，包含工具处理器（如 get_transaction、hybrid_search_similar_frauds 等），这些处理器在主机端运行 pymongo 查询，并处理 AP2 授权验证。MongoDB Atlas 层存储交易数据、决策记录、审计事件和授权收据，支持向量索引、全文索引和图遍历。设计决策包括：路径 A（主机端自定义工具）作为默认，确保凭据安全；使用 $rankFusion 实现混合搜索，减少客户端合并；通过 requires_action 实现人工审批门控，保证高风险决策可控。

完整实战教程

本教程演示如何将 MongoDB Atlas 集成到运行在 Claude 托管代理 (CMA) 上的代理中——作为其检索引擎、图存储和记录系统——仅使用标准的 CMA 模式（自定义工具和 MCP 工具集），无需平台级 MongoDB 集成。可运行的辅助代码位于此笔记本旁的 mongodb_on_cma/ 目录中。每个部分都链接到完整文件，以便叙述聚焦于 *MongoDB 的接入点* 和 *代理的价值所在*。

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

代理的价值所在： CMA 在 Anthropic 端运行代理循环和沙箱。您的应用程序拥有通往 MongoDB 的 数据路径。代理对 MongoDB 检索到的内容进行推理，通过 CMA 原生的 requires_action 门控在风险调用时暂停等待人工，并写回其裁决——所有这些都不需要数据库凭据进入代理上下文或其沙箱。

这里的 MongoDB 是您的 操作数据存储和检索引擎——您的记录系统，由您的应用程序控制。这与 CMA 在平台上原生管理的内存功能不同。两者是互补的。

最终您将能够：

示例是一个 人在环路的欺诈审查代理，但这些模式是垂直无关的：更换集合和工具，相同的结构即可服务于支持、研究或运营代理。

本指南涵盖的内容

  1. 将 MongoDB 连接到托管代理
  2. 检索：四种模式，一个引擎
  3. 端到端代理：人在环路的欺诈审查
  4. MongoDB Atlas 作为记录系统和审计支柱
  5. 回顾
%%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？

library-RAG 笔记本 首先介绍了 $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 或更高版本，因此默认满足此要求。

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 索引文档 涵盖了索引设置。

教学代码内联在本 notebook 中：四个检索管道构建器（第 2 节）以及自定义工具处理器和 requires_action 门控循环（第 3 节）。notebook 旁边的 mongodb_on_cma/ 包仅包含你导入而非阅读的设置样板：config.py（索引名称 + 可调参数）、embeddings.py（Atlas/Voyage 客户端）、tools.py（Atlas 种子/索引设置 + 共享决策/审计文档整形器）、ap2_mandates.py（AP2 签名/验证——通过 verify_mandates 调用的加密黑盒），以及 seed.py（示例数据加载器）。导入它们，连接客户端，并解析运行配置。

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'}"
)

1. 将 MongoDB 连接到托管代理

代理循环及其沙箱由 Anthropic 托管——这就是“托管”的含义。你唯一托管的是通往 MongoDB 的数据路径。在每条路径上，MongoDB 凭据都位于边界的*你*这一侧：它永远不会进入代理上下文、云沙箱环境或代理可读取的文件。

为什么本食谱通过 client.beta.* 与 CMA 通信，而不是使用 Claude Agent SDK。
托管代理 (CMA) 是一个托管 REST API：Anthropic 运行代理循环和沙箱，你通过标准 Anthropic SDK 的 beta 接口（client.beta.agents、client.beta.sessions、client.beta.sessions.events）创建代理/会话并流式传输事件来驱动它。没有单独的“CMA SDK”——这就是使用 CMA 的方式。Claude Agent SDK（claude-agent-sdk，在仓库的 claude_agent_sdk/ 食谱中展示）是一个不同的产品：一个在你自己的进程中、在你自己的基础设施上运行代理循环的库。两者是互补的——你可以使用 Agent 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 节将此确切模式扩展到五个工具和真实工作负载。

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}")

打开事件流，提出一个问题，并驱动往返过程：在每次

agent.custom_tool_use 时，使用 pymongo 在主机端运行查询，并回传一个

user.custom_tool_result。结果通过 {"_id": 0} 进行投影，因为原始的 MongoDB

ObjectId 无法进行 JSON 序列化。

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")

其他选项（简要）

路径 A 是默认选项，本手册其余部分均使用它。对于不适合的情况，存在两种替代方案——你不需要它们来继续阅读：

权衡一句话：路径 A 保持查询集固定，并将秘密保存在你的后端（最安全的默认值）；路径 B 将直接访问权交给沙箱；路径 C 给代理最广泛的查询表面，但代价是运行服务器。如有疑问，请坚持使用路径 A。

2. 检索：四种模式，一个引擎

一个代理的可靠性取决于其检索能力。以下是四种 MongoDB 检索模式单独展示——这里是构建器，这里是它返回的内容——以便您可以将单一模式提取到自己的代理工具中。每个都是一个普通的聚合管道构建器；第 3 节的代理通过其工具调用这些相同的函数。

| 模式 | 构建器 | MongoDB 阶段 |

| --- | --- | --- |

| 向量搜索 | build_vector_pipeline | $vectorSearch |

| 全文搜索 | build_lexical_pipeline | $search |

| 混合（倒数排名融合） | build_rank_fusion_pipeline | $rankFusion (8.0+) |

| 图遍历 | build_graph_pipeline | $graphLookup |

构建器在下一个单元格中内联定义——每个返回一个聚合阶段列表，您可以将其提取到自己的集合中。索引名称和投影字段来自 config.py；EMBED_DIM 和索引常量在设置单元格中导入。

首先，从 example_data/mongodb_on_cma/ 加载种子数据，并构建索引。该数据已预计算嵌入，因此无需提供程序密钥即可运行。

# 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)")

模式 1：向量搜索（$vectorSearch）

语义召回：查找其 embedding 与查询向量最接近的文档。此示例复用现有文档的嵌入作为查询（一个待处理案件寻找已判决的先例），因此无需调用嵌入 API。构建器通过 status 字段过滤出已判决的案件。

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]}")

模式 2：全文搜索（$search，BM25）

对文本字段进行关键词和短语匹配——这能捕捉到嵌入向量模糊化的确切名称、ID 和代码。

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]}")

模式 3：混合（倒数排名融合）

将向量和全文排名融合到一个结果集中。$rankFusion（MongoDB 8.0+）在服务器端运行两个输入管道，并通过倒数排名合并它们——无需客户端合并，无需第二次往返。

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]}")

模式 4：图遍历（$graphLookup）

沿着 sender.account_number -> recipient.account_number 链接来揭示一个网络（这里是一个循环资金流环）。这是一个*关系*信号，特意与上面的相似度排名分开。

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']}"
)

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

3. 端到端智能体：人在回路的欺诈审查

现在所有组件协同工作。一个托管智能体审查被标记的交易，将每条建议基于第2节中的 MongoDB 信号（通过 自定义工具 在路径 A 的往返中暴露），并在高风险案例上暂停等待人工介入。决策和仅追加的审计跟踪被写回同一集群。

<p align="center">

代理循环在 Anthropic 上运行；数据路径在您的笔记本中运行。因此，MONGO_URI 和嵌入密钥永远不会进入代理或其沙箱。人工等待是持久的——会话通过 requires_action 在服务器端空闲，直到您响应——而下面的简单流式循环仅用于开发（在生产环境中，请使用 webhook 模式从持久后端驱动它）。

AP2 强制要求：先验证，再推理

此代理审查*代理发起*的付款，因此在进行任何行为分析之前，它会验证 AP2（代理支付协议） 授权——证明用户授权付款的加密签名凭证。AP2 是一个用于代理驱动商务的开放协议。其授权（结账授权和支付授权）是防篡改的签名数字对象，形成可审计的链。您无需学习该协议即可遵循本指南：调用 verify_mandates 工具并根据其裁决（有效、满足约束、双重支付）采取行动。签名和验证细节（ES256 JWT）封装在 ap2_mandates.py 中，签名授权和收据与交易一起存储在 MongoDB 中。

下面的单元格扮演“可信表面”（生成密钥对，将签名授权附加到待处理案例）。私钥保留在本地 Python 状态中——绝不会出现在代理、工具结果或数据库中。

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]}…")

自定义工具与主机端处理器

每个工具都是 type: "custom" 类型的工具。当代理调用其中一个工具时，会话会暂停，本笔记本会使用 pymongo 运行处理器，并将 JSON 结果发送回去——代理永远不会接触数据库或连接字符串。处理器实现定义在下一个单元格中：它们包装了第 2 节的构建器以及 AP2 验证（verify_mandates）。

HANDLERS 仅映射五个数据工具；escalate 被故意排除在外，因为门控循环将其路由到人工解析器而不是处理器（参见下一节）。

# 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,
}

创建代理、环境和会话

model、system 和 tools 都存在于代理上。会话引用代理并配置沙箱。网络是受限的：代理仅通过主机端的往返访问 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}")

运行带人工门控的审查

打开事件流，发送待处理交易队列，并使用 run_gate_loop 驱动门控循环：数据工具调用会自动从 HANDLERS 中处理，而 escalate 调用会暂停等待人工介入。当设置了 AUTO_APPROVE（例如在 CI 中），门控会确定性地解析，并在种子结构化案例上覆盖代理，从而显示不同的人工裁决。否则，系统会提示你进行内联操作。

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")

审查决策与审计轨迹

从 MongoDB 中读回决策和审计轨迹。决策和审计集合是仅追加的，因此读回仅限于本次运行。收尾断言使其成为自检：如果重新运行未能复现预期的车道混合，则会大声失败。

人工覆盖计数反映了门控的解决方式：使用 AUTO_APPROVE 时，计数为 1（显示的是在 txn-review-struct 上植入的覆盖），而在交互式运行中，当你对每次升级都同意代理的建议时，计数为 0——这两种情况都是预期的。

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")

4. MongoDB Atlas 作为记录系统与审计支柱

退一步看运行状态存储在哪里：代理所做的一切都持久化在它检索数据的同一个集群中。这就是写入端单引擎的回报——transactions 是操作记录（每个案例推进 pending → approved 或 rejected），transaction_decisions 保存不可变的裁决，audit_events 是仅追加的追踪，而 mandate_receipts 支持双重支出检测。一个案例，端到端：

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']}"
        )

生产环境的 Webhooks（此处仅作提示，不运行）： 上述流式循环非常适合开发，但无法在重启后继续。对于生产环境，请为 session.status_idled 注册一个 webhook：你的处理程序将案例排队等待审核员处理，稍后 POST user.custom_tool_result。data-tool 处理程序完全相同；只有触发器发生变化。参见 CMA_operate_in_production.ipynb。

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")

5. 回顾

你已将 MongoDB Atlas 连接到 Claude 托管代理，并在一次构建中将其从连接提升为记录系统：

集成面很小：自定义工具背后的几行 pymongo。交换集合和工具，同样的形状可将任何代理锚定在 MongoDB 中。检索构建器、工具处理程序和门控循环都在此笔记本中；mongodb_on_cma/ 包仅包含设置样板。

常见问题

常见错误包括：1) 未设置 MONGO_URI 或 Anthropic 凭据，导致连接失败，需检查 .env 文件；2) 索引未创建或未同步，导致检索失败，需运行 ensure_indexes 并等待索引就绪；3) MongoDB 版本低于 8.0 导致 $rankFusion 不可用，需使用 Atlas 集群或升级；4) ObjectId 无法 JSON 序列化，需投影掉 _id 字段；5) 代理工具调用超时或未响应，需检查事件流处理逻辑；6) 人工审批门控未触发，需确保 escalate 工具被正确调用。排查方法：使用 preflight 检查环境，查看日志输出，验证索引状态，测试单个工具处理器。生产优化：使用 webhook 模式替代流式循环，实现持久化处理；启用重排序（rerank）提升检索质量；使用 AUTO_APPROVE 进行 CI 测试；考虑使用自托管沙箱或 MCP 服务器以扩展查询能力。

企业应用扩展

本教程的模式可扩展到其他领域，如客户支持、研究助手或运营代理。开发者可以替换集合和工具定义，复用相同的架构。进一步优化包括：1) 使用 MongoDB Atlas 的流式变更跟踪（change streams）实现实时数据同步；2) 集成更多检索模式，如使用 $search 的自动补全或拼写纠正；3) 利用 MongoDB 的聚合管道进行复杂特征工程，为代理提供更丰富的上下文；4) 结合 MongoDB Atlas 的在线归档和分片功能，处理大规模数据；5) 实现更精细的权限控制，通过数据库用户角色限制代理可访问的数据范围。

总结

本教程面向有编程基础的开发者，深入讲解如何将 MongoDB Atlas 集成到 Claude 托管代理（CMA）中，作为检索引擎、图存储和记录系统。通过主机端自定义工具实现凭据安全的数据路径，利用 MongoDB 的向量搜索、全文搜索、混合搜索（RRF）和图遍历四种检索模式，构建一个带人工审批门控的欺诈审查代理。教程涵盖环境准备、核心原理、架构设计、代码分析、常见错误排查及生产优化，帮助开发者掌握 AI 工程实践中的关键集成技术。

内容来源与验证

继续系统学习

站内还有更多完整实战教程

加入会员，解锁全部教程、完整代码与持续更新内容。

加入会员
查看全部AI 智能体教程