{ "cells": [ { "cell_type": "markdown", "metadata": {}, "source": [ "# Fraud Review Agent with MongoDB Atlas and Claude Managed Agents\n", "\n", "This cookbook shows how to bring **MongoDB Atlas** to an agent running on\n", "[Claude Managed Agents](README.md) (CMA) — as its retrieval engine, its graph store, and its\n", "system of record — using only the standard CMA patterns (custom tools and MCP toolsets), with\n", "**no platform-level MongoDB integration required**. The runnable helpers live beside this\n", "notebook in [`mongodb_on_cma/`](mongodb_on_cma/). Each section links the full file so the\n", "narrative stays focused on *where MongoDB plugs in* and *where the agent adds value*.\n", "\n", "**Where MongoDB Plugs In:** Most agent stacks bolt together three or four systems: a vector\n", "database for semantic search, a separate engine for keywords, a graph store for relationships,\n", "and an operational database for the records themselves. Every seam is another integration,\n", "another credential, another place the agent's view of the world can drift. MongoDB collapses\n", "that into one engine: the same documents are searchable by **vector** (`$vectorSearch`), by\n", "**full-text** (`$search`), as a **hybrid** of the two fused with reciprocal rank fusion, or RRF\n", "(`$rankFusion`), and traversable as a **graph** (`$graphLookup`) — and they are the same\n", "documents the agent reads, writes, and persists its decisions to. One query language, one\n", "cluster, one connection (`pymongo`).\n", "\n", "**Where the Agent Adds Value:** CMA runs the agent loop and sandbox on Anthropic's side. Your\n", "application owns the **data path** to MongoDB. The agent reasons over what MongoDB retrieves,\n", "pauses for a human on risky calls through CMA's native `requires_action` gate, and writes its\n", "verdict back — all without the database credential ever entering the agent context or its\n", "sandbox.\n", "\n", "> MongoDB here is your **operational data store and retrieval engine** — your system of record,\n", "> controlled by your application. That's distinct from the memory features CMA manages natively\n", "> on the platform. The two are complementary.\n", "\n", "**By the end you will be able to:**\n", "\n", "- Connect a credential-safe MongoDB Atlas data path into a Claude Managed Agent in three ways.\n", "- Lift vector, full-text, hybrid, and graph retrieval into an agent's custom tools.\n", "- Gate risky agent decisions behind CMA's native human-in-the-loop pause.\n", "- Make one MongoDB Atlas cluster the agent's system of record and audit backbone.\n", "\n", "The worked example is a **human-in-the-loop fraud-review agent**, but the patterns are\n", "vertical-agnostic: swap the collection and the tools, and the same shape serves a support,\n", "research, or operations agent.\n", "\n", "## What this guide covers\n", "\n", "1. [Connect MongoDB to a managed agent](#1-connect-mongodb-to-a-managed-agent)\n", "2. [Retrieve: four patterns, one engine](#2-retrieve-four-patterns-one-engine)\n", "3. [The end-to-end agent: human-in-the-loop fraud review](#3-the-end-to-end-agent-human-in-the-loop-fraud-review)\n", "4. [MongoDB Atlas as the system of record and audit backbone](#4-mongodb-atlas-as-the-system-of-record-and-audit-backbone)\n", "5. [Recap](#5-recap)" ] }, { "cell_type": "code", "execution_count": 1, "metadata": {}, "outputs": [], "source": [ "%%capture\n", "# Third-party dependencies: the Anthropic SDK (drives Claude Managed Agents via client.beta.*),\n", "# pymongo (the host-side data path), python-dotenv, and cryptography + pyjwt (sign/verify the\n", "# AP2 mandates). This installs the PyPI packages only — the notebook also imports the adjacent\n", "# `mongodb_on_cma/` package and `utilities.py`, so run it from a clone of this repo (see Setup).\n", "%pip install -q \"anthropic>=0.109.0\" pymongo python-dotenv cryptography pyjwt" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## Prerequisites\n", "\n", "**Required knowledge:** Python and `pymongo` basics, plus passing familiarity with the Claude\n", "API. New to MongoDB + Claude? The\n", "[library-RAG notebook](../third_party/MongoDB/rag_using_mongodb.ipynb) covers `$vectorSearch`\n", "first. New to the custom-tool gate?\n", "[`CMA_gate_human_in_the_loop.ipynb`](CMA_gate_human_in_the_loop.ipynb) teaches the round-trip.\n", "\n", "**Required tools:** Python 3.11+, an [Anthropic API key](https://console.anthropic.com), and a\n", "MongoDB Atlas cluster.\n", "\n", "## Setup\n", "\n", "**Required:** `MONGO_URI` (an Atlas SRV connection string) plus Anthropic API access. A **free\n", "M0 cluster runs everything in this cookbook** — vector, full-text, hybrid `$rankFusion`, and\n", "graph traversal — so there is no paid-tier requirement. Hybrid search uses the native\n", "`$rankFusion` stage, which needs MongoDB 8.0+; every current Atlas cluster (M0 included) is on\n", "8.0 or later, so this holds by default.\n", "\n", "```bash\n", "uv sync --all-extras # from the repo root\n", "cp .env.example .env # then add MONGO_URI (and Anthropic auth, if not already configured)\n", "```\n", "\n", "**Optional embeddings/rerank provider:** The seed fixture ships precomputed embeddings, so the\n", "cookbook runs without one. To enable the live-embedding and reranker paths, set **one** of\n", "`MDB_ATLAS_API_KEY` (the MongoDB Atlas AI endpoint, serving both `/v1/embeddings` and\n", "`/v1/rerank`) or `VOYAGE_API_KEY` (the `voyageai` SDK). `ENABLE_RERANK=1` adds the reranker\n", "second stage. `AUTO_APPROVE=1` resolves the human gate deterministically for CI. `COOKBOOK_MODEL`\n", "overrides the agent model (default `claude-haiku-4-5`).\n", "\n", "**No Atlas cluster yet?** The partner notebook [`rag_using_mongodb.ipynb`](../third_party/MongoDB/rag_using_mongodb.ipynb) walks through creating a free cluster and getting your `MONGO_URI`; the [Atlas Search index docs](https://www.mongodb.com/docs/atlas/atlas-search/create-index/) cover index setup." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "The teaching code lives **inline in this notebook**: the four retrieval pipeline builders\n", "(Section 2) and the custom-tool handlers plus the `requires_action` gate loop (Section 3). The\n", "[`mongodb_on_cma/`](mongodb_on_cma/) package next to the notebook holds only setup boilerplate\n", "you import rather than read: [`config.py`](mongodb_on_cma/config.py) (index names + tunables),\n", "[`embeddings.py`](mongodb_on_cma/embeddings.py) (the Atlas/Voyage client),\n", "[`tools.py`](mongodb_on_cma/tools.py) (Atlas seed/index setup + shared decision/audit doc\n", "shapers), [`ap2_mandates.py`](mongodb_on_cma/ap2_mandates.py) (AP2 signing/verification — a\n", "crypto black box you call through `verify_mandates`), and\n", "[`seed.py`](mongodb_on_cma/seed.py) (the example-data loader). Import them, connect the clients,\n", "and resolve the run configuration." ] }, { "cell_type": "code", "execution_count": 2, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "warning: no ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN / ANTHROPIC_PROFILE found; relying on the SDK / `ant` CLI to resolve credentials.\n", "model=claude-haiku-4-5 rerank=on provider=yes\n" ] } ], "source": [ "import hashlib\n", "import json\n", "import logging\n", "import os\n", "from datetime import UTC, datetime\n", "from typing import Any\n", "\n", "import dotenv\n", "from anthropic import Anthropic\n", "from mongodb_on_cma import (\n", " EMBED_DIM,\n", " build_audit_event,\n", " build_decision_doc,\n", " ensure_indexes,\n", " has_anthropic_auth,\n", " load_seed,\n", " make_embedding_client,\n", " missing_required_env,\n", " preflight,\n", " prepare_seed,\n", " rerank,\n", " resolve_model,\n", " seed_collection,\n", " server_version,\n", " supports_rank_fusion,\n", ")\n", "from mongodb_on_cma.ap2_mandates import (\n", " attach_mandates,\n", " store_mandate_receipt,\n", " tool_verify_mandates,\n", ")\n", "from pymongo import MongoClient\n", "from utilities import wait_for_idle_status\n", "\n", "dotenv.load_dotenv()\n", "\n", "missing = missing_required_env()\n", "assert not missing, f\"Set these in your environment / .env: {missing}\"\n", "# Anthropic auth is a soft check: an API key, an auth token, or a profile (e.g. the `ant`\n", "# CLI's workload-identity federation) all work, so warn rather than block if none is set.\n", "if not has_anthropic_auth():\n", " print(\n", " \"warning: no ANTHROPIC_API_KEY / ANTHROPIC_AUTH_TOKEN / ANTHROPIC_PROFILE found; \"\n", " \"relying on the SDK / `ant` CLI to resolve credentials.\"\n", " )\n", "\n", "MODEL = resolve_model()\n", "ENABLE_RERANK = os.getenv(\"ENABLE_RERANK\", \"\").lower() in (\"1\", \"true\")\n", "AUTO_APPROVE = os.getenv(\"AUTO_APPROVE\", \"\").lower() in (\"1\", \"true\")\n", "\n", "# Quiet the SDK's one-shot notice that ANTHROPIC_API_KEY shadows profile/federation\n", "# auto-discovery — expected when an API key is set; harmless under `ant` / WIF auth.\n", "logging.getLogger(\"anthropic.lib.credentials._auth\").setLevel(logging.ERROR)\n", "client = Anthropic()\n", "mongo = MongoClient(os.environ[\"MONGO_URI\"])\n", "db = mongo[\"fraud_review_demo\"]\n", "coll = db[\"transactions\"]\n", "ai_client = make_embedding_client() # Atlas / Voyage / None (seed ships precomputed vectors)\n", "\n", "print(\n", " f\"model={MODEL} rerank={'on' if ENABLE_RERANK else 'off'} provider={'yes' if ai_client else 'none'}\"\n", ")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 1. Connect MongoDB to a managed agent\n", "\n", "The agent loop and its sandbox are **Anthropic-hosted** — that is what \"Managed\" means. The\n", "only thing you host is the **data path** to MongoDB. On every path, the MongoDB credential\n", "lives on *your* side of the boundary: it never enters the agent context, a cloud sandbox's\n", "environment, or a file the agent can read.\n", "\n", "> **Why this cookbook talks to CMA through `client.beta.*` and not the Claude Agent SDK.**\n", "> Managed Agents (CMA) is a [hosted REST API](https://platform.claude.com/docs/en/managed-agents/overview):\n", "> Anthropic runs the agent loop and the sandbox, and you drive it by creating agents/sessions\n", "> and streaming events through the standard Anthropic SDK's beta surface\n", "> (`client.beta.agents`, `client.beta.sessions`, `client.beta.sessions.events`). There is no\n", "> separate \"CMA SDK\" — this *is* how you use CMA. The\n", "> [Claude Agent SDK](https://code.claude.com/docs/en/agent-sdk/overview)\n", "> (`claude-agent-sdk`, showcased in the repo's\n", "> [`claude_agent_sdk/`](../claude_agent_sdk/) cookbooks) is a different product: a library that\n", "> runs the agent loop **inside your own process**, on your own infrastructure. The two are\n", "> complementary — you'd prototype locally with the Agent SDK and run in production on CMA — but\n", "> a CMA integration like this one is built on the beta sessions/events API. That choice is also\n", "> what makes the credential boundary below work: because *your* process handles each custom-tool\n", "> call, the MongoDB secret stays host-side (see Path A).\n", "\n", "There are three ways to wire the data path. **This cookbook uses Path A throughout** — it is\n", "the recommended, lightest pattern and the one that keeps the credential fully host-side. Paths\n", "B and C are summarized after it for completeness.\n", "\n", "### Path A: host-side custom tool (recommended)\n", "\n", "Your application defines a `custom` tool. When the agent calls it, the session pauses with\n", "`requires_action`, **your** process runs the query with `pymongo`, and you post back a\n", "`user.custom_tool_result`. The secret never enters the sandbox, MongoDB can sit in a private\n", "VPC, and the agent is constrained to a fixed, audited set of queries. It is the lightest path —\n", "no standing server — and Anthropic's recommended pattern for any secret-bearing data source.\n", "\n", "To see the wiring unobscured, here is the whole round-trip with one tiny `find_notes` tool over\n", "a throwaway `notes` collection. **This is the core integration: a few lines of `pymongo`, run\n", "host-side, behind a custom tool.** Section 3 scales this exact shape to five tools and a real\n", "workload." ] }, { "cell_type": "code", "execution_count": 3, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "seeded 3 notes; session: sesn_016ufm58BVJ8aZeWjUt39bsq\n" ] } ], "source": [ "notes = mongo[\"cookbook_mongodb_on_cma\"][\"notes\"]\n", "notes.delete_many({})\n", "notes.insert_many(\n", " [\n", " {\"note_id\": \"n1\", \"tag\": \"todo\", \"text\": \"Renew the SSL certificate before it expires.\"},\n", " {\"note_id\": \"n2\", \"tag\": \"idea\", \"text\": \"Add a dark-mode toggle to the dashboard.\"},\n", " {\"note_id\": \"n3\", \"tag\": \"todo\", \"text\": \"Email the quarterly report to the finance team.\"},\n", " ]\n", ")\n", "\n", "notes_agent = client.beta.agents.create(\n", " name=\"cookbook-mongodb-notes\",\n", " model=MODEL,\n", " system=(\n", " \"You answer questions about the user's notes. Use the find_notes tool to fetch notes \"\n", " \"(optionally filtered by `tag`) before answering, then give a concise final answer.\"\n", " ),\n", " tools=[\n", " {\n", " \"type\": \"agent_toolset_20260401\",\n", " \"default_config\": {\"enabled\": True, \"permission_policy\": {\"type\": \"always_allow\"}},\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"find_notes\",\n", " \"description\": \"Return stored notes, optionally filtered by tag (e.g. 'todo', 'idea').\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\"tag\": {\"type\": \"string\", \"description\": \"optional tag filter\"}},\n", " \"required\": [],\n", " },\n", " },\n", " ],\n", ")\n", "notes_env = client.beta.environments.create(\n", " name=\"cookbook-mongodb-notes-env\", config={\"type\": \"cloud\", \"networking\": {\"type\": \"limited\"}}\n", ")\n", "notes_session = client.beta.sessions.create(\n", " environment_id=notes_env.id,\n", " agent={\"type\": \"agent\", \"id\": notes_agent.id, \"version\": notes_agent.version},\n", " title=\"MongoDB notes (host-side custom tool)\",\n", ")\n", "print(f\"seeded {notes.count_documents({})} notes; session: {notes_session.id}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "Open the event stream, ask a question, and drive the round-trip: on each\n", "`agent.custom_tool_use`, run the query **host-side** with `pymongo` and post back a\n", "`user.custom_tool_result`. Results are projected with `{\"_id\": 0}` because a raw MongoDB\n", "`ObjectId` is not JSON-serializable." ] }, { "cell_type": "code", "execution_count": 4, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "host-side find_notes round-trips:\n", " find_notes({'tag': 'todo'}) -> 2 notes\n", "archived the demo agent / environment / session; dropped the notes collection\n" ] } ], "source": [ "def run_find_notes(args: dict) -> dict:\n", " \"\"\"Host-side handler: pymongo runs here; _id (ObjectId) is projected out so the result is JSON-able.\"\"\"\n", " flt = {\"tag\": args[\"tag\"]} if args.get(\"tag\") else {}\n", " return {\"notes\": list(notes.find(flt, {\"_id\": 0}))}\n", "\n", "\n", "tool_use_events, responded, round_trips = {}, set(), []\n", "question = \"Using find_notes, list every note tagged 'todo', then tell me how many there are.\"\n", "\n", "with client.beta.sessions.events.stream(notes_session.id) as stream:\n", " client.beta.sessions.events.send(\n", " session_id=notes_session.id,\n", " events=[{\"type\": \"user.message\", \"content\": [{\"type\": \"text\", \"text\": question}]}],\n", " )\n", " for ev in stream:\n", " if ev.type == \"agent.custom_tool_use\":\n", " tool_use_events[ev.id] = ev\n", " elif ev.type == \"session.status_idle\" and ev.stop_reason:\n", " if ev.stop_reason.type == \"requires_action\":\n", " for event_id in ev.stop_reason.event_ids:\n", " if event_id in responded:\n", " continue\n", " tev = tool_use_events[event_id]\n", " result = (\n", " run_find_notes(tev.input)\n", " if tev.name == \"find_notes\"\n", " else {\"error\": \"unknown\"}\n", " )\n", " if tev.name == \"find_notes\":\n", " round_trips.append(\n", " {\"tool_input\": tev.input, \"returned\": len(result[\"notes\"])}\n", " )\n", " client.beta.sessions.events.send(\n", " session_id=notes_session.id,\n", " events=[\n", " {\n", " \"type\": \"user.custom_tool_result\",\n", " \"custom_tool_use_id\": event_id,\n", " \"content\": [{\"type\": \"text\", \"text\": json.dumps(result)}],\n", " }\n", " ],\n", " )\n", " responded.add(event_id)\n", " elif ev.stop_reason.type == \"end_turn\":\n", " break\n", " elif ev.type == \"session.status_terminated\":\n", " break\n", "\n", "wait_for_idle_status(client, notes_session.id)\n", "print(\"host-side find_notes round-trips:\")\n", "for rt in round_trips:\n", " print(f\" find_notes({rt['tool_input']}) -> {rt['returned']} notes\")\n", "\n", "client.beta.sessions.archive(notes_session.id)\n", "client.beta.environments.archive(notes_env.id)\n", "client.beta.agents.archive(notes_agent.id)\n", "notes.drop()\n", "print(\"archived the demo agent / environment / session; dropped the notes collection\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Other options (brief)\n", "\n", "Path A is the default and the rest of this cookbook uses it. Two alternatives exist for cases\n", "it doesn't fit — you don't need them to follow along:\n", "\n", "- **Path B — in-sandbox `pymongo` (self-hosted sandbox).** If you run the sandbox on your own\n", " infrastructure, the MongoDB client can live *inside* it and the agent queries MongoDB straight\n", " from its `bash` tool, with `MONGO_URI` as an ordinary container env var. Pick this only for\n", " self-hosted sandboxes (on a cloud sandbox the connection string would land in session history).\n", " Runnable image: [`self_hosted_sandboxes/docker/`](self_hosted_sandboxes/docker/).\n", "- **Path C — self-hosted MongoDB MCP server.** Host the official MongoDB MCP server behind HTTPS\n", " and register it as an `mcp_toolset` so the agent can query the full surface (`find`,\n", " `aggregate`, `$vectorSearch`) rather than a fixed query set. Standard CMA MCP + vault wiring.\n", "\n", "The trade-off in one line: **Path A** keeps the query set fixed and the secret in your backend\n", "(the safest default); **Path B** hands the sandbox direct access; **Path C** gives the agent the\n", "broadest query surface at the cost of running a server. When in doubt, stay on Path A." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 2. Retrieve: four patterns, one engine\n", "\n", "An agent is only as well-grounded as its retrieval. These are the four MongoDB retrieval\n", "patterns shown **in isolation** — here's the builder, here's what it returns — so you can lift a\n", "single pattern into your own agent's tools. Each is a plain aggregation-pipeline builder; the\n", "Section 3 agent calls these same functions through its tools.\n", "\n", "| Pattern | Builder | MongoDB stage |\n", "| --- | --- | --- |\n", "| Vector search | `build_vector_pipeline` | `$vectorSearch` |\n", "| Full-text search | `build_lexical_pipeline` | `$search` |\n", "| Hybrid (reciprocal rank fusion) | `build_rank_fusion_pipeline` | `$rankFusion` (8.0+) |\n", "| Graph traversal | `build_graph_pipeline` | `$graphLookup` |\n", "\n", "The builders are defined inline in the next cell — each returns a list of aggregation stages you\n", "can lift into your own collection. Index names and the projected fields come from\n", "[`config.py`](mongodb_on_cma/config.py); `EMBED_DIM` and the index constants are imported in the\n", "setup cell above.\n", "\n", "First, load the seed fixture from\n", "[`example_data/mongodb_on_cma/`](example_data/mongodb_on_cma/seed_transactions.jsonl) and build\n", "the indexes. The fixture ships precomputed embeddings, so this runs with no provider key." ] }, { "cell_type": "code", "execution_count": 5, "metadata": {}, "outputs": [], "source": [ "# The four retrieval builders — each returns a plain aggregation pipeline you can lift into your\n", "# own collection. These are the functions Section 3's agent calls through its custom tools.\n", "from mongodb_on_cma.config import (\n", " DECIDED_STATUSES,\n", " LEXICAL_PATHS,\n", " PROJECT_FIELDS,\n", " SEARCH_INDEX_NAME,\n", " VECTOR_INDEX_NAME,\n", ")\n", "\n", "\n", "def _project_stage(with_score: bool = False) -> dict:\n", " proj: dict[str, Any] = {f: 1 for f in PROJECT_FIELDS}\n", " proj[\"_id\"] = 0\n", " if with_score:\n", " proj[\"score\"] = {\"$meta\": \"score\"}\n", " return {\"$project\": proj}\n", "\n", "\n", "def build_vector_pipeline(\n", " qvec, *, limit, candidates=None, vector_index=VECTOR_INDEX_NAME, status_in=DECIDED_STATUSES\n", ") -> list[dict]:\n", " return [\n", " {\n", " \"$vectorSearch\": {\n", " \"index\": vector_index,\n", " \"path\": \"embedding\",\n", " \"queryVector\": qvec,\n", " \"numCandidates\": candidates or max(50, limit * 10),\n", " \"limit\": limit,\n", " \"filter\": {\"status\": {\"$in\": list(status_in)}},\n", " }\n", " },\n", " _project_stage(),\n", " ]\n", "\n", "\n", "def build_lexical_pipeline(query, *, limit, search_index=SEARCH_INDEX_NAME) -> list[dict]:\n", " return [\n", " {\"$search\": {\"index\": search_index, \"text\": {\"query\": query, \"path\": LEXICAL_PATHS}}},\n", " {\"$limit\": limit},\n", " _project_stage(),\n", " ]\n", "\n", "\n", "def build_rank_fusion_pipeline(\n", " qvec,\n", " query,\n", " *,\n", " k,\n", " vector_index=VECTOR_INDEX_NAME,\n", " search_index=SEARCH_INDEX_NAME,\n", " status_in=DECIDED_STATUSES,\n", ") -> list[dict]:\n", " # $rankFusion (MongoDB 8.0+) runs both input pipelines and fuses them by reciprocal rank —\n", " # one aggregation, server-side. Each input pipeline gets weight 1 (uniform). To bias toward\n", " # semantic vs. lexical, add \"combination\": {\"weights\": {\"vector\": w, \"lexical\": w}}.\n", " candidates = max(50, k * 10)\n", " per_branch = max(k * 4, 20)\n", " return [\n", " {\n", " \"$rankFusion\": {\n", " \"input\": {\n", " \"pipelines\": {\n", " \"vector\": [\n", " {\n", " \"$vectorSearch\": {\n", " \"index\": vector_index,\n", " \"path\": \"embedding\",\n", " \"queryVector\": qvec,\n", " \"numCandidates\": candidates,\n", " \"limit\": per_branch,\n", " \"filter\": {\"status\": {\"$in\": list(status_in)}},\n", " }\n", " }\n", " ],\n", " \"lexical\": [\n", " {\n", " \"$search\": {\n", " \"index\": search_index,\n", " \"text\": {\"query\": query, \"path\": LEXICAL_PATHS},\n", " }\n", " },\n", " {\"$limit\": per_branch},\n", " ],\n", " }\n", " },\n", " }\n", " },\n", " {\"$limit\": k},\n", " _project_stage(with_score=True),\n", " ]\n", "\n", "\n", "def build_graph_pipeline(\n", " account_id: str, *, max_depth: int = 4, collection: str = \"transactions\"\n", ") -> list[dict]:\n", " return [\n", " {\"$match\": {\"sender.account_number\": account_id}},\n", " {\n", " \"$graphLookup\": {\n", " \"from\": collection,\n", " \"startWith\": \"$recipient.account_number\",\n", " \"connectFromField\": \"recipient.account_number\",\n", " \"connectToField\": \"sender.account_number\",\n", " \"as\": \"chain\",\n", " \"maxDepth\": max_depth,\n", " \"depthField\": \"depth\",\n", " }\n", " },\n", " ]\n", "\n", "\n", "def summarize_ring(graph_doc: dict, *, seed_account: str) -> dict:\n", " # Turn a $graphLookup chain into fraud-ring signals: circular flow back to the seed account,\n", " # layering (many small transfers), and overall network size.\n", " chain: list[dict] = list(graph_doc.get(\"chain\", []))\n", " accounts: set[str] = set()\n", " small_transfers = 0\n", " circular_flow = False\n", " for edge in chain:\n", " sender = (edge.get(\"sender\") or {}).get(\"account_number\")\n", " recipient = (edge.get(\"recipient\") or {}).get(\"account_number\")\n", " accounts.update(a for a in (sender, recipient) if a)\n", " if recipient == seed_account:\n", " circular_flow = True\n", " if float(edge.get(\"amount\", 0) or 0) < 1000:\n", " small_transfers += 1\n", " network_size = len(chain)\n", " layering = small_transfers >= 5\n", " return {\n", " \"network_size\": network_size,\n", " \"unique_accounts\": len(accounts),\n", " \"circular_flow\": circular_flow,\n", " \"layering\": layering,\n", " \"suspicious_patterns\": circular_flow or layering or network_size >= 3,\n", " }" ] }, { "cell_type": "code", "execution_count": 6, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "seeded 20 transactions\n", "server=8.0.27 hybrid search=$rankFusion (native)\n" ] } ], "source": [ "docs = prepare_seed(load_seed(), now=datetime.now(UTC))\n", "count = seed_collection(coll, docs)\n", "print(f\"seeded {count} transactions\")\n", "\n", "# Create the vector + Atlas Search indexes if absent, then block until both are queryable AND\n", "# actually reflect the freshly-seeded docs (Atlas Search is eventually consistent).\n", "ensure_indexes(coll, dim=EMBED_DIM)\n", "\n", "check = preflight(coll)\n", "for issue in check[\"issues\"]:\n", " print(\"PREFLIGHT:\", issue)\n", "assert check[\"ok\"], \"Fix the preflight issues above before continuing.\"\n", "\n", "# This cookbook uses the native `$rankFusion` stage for hybrid search, which needs MongoDB 8.0+.\n", "# Every current Atlas cluster — including the free M0 tier — runs 8.0 or later, so this holds by\n", "# default; the check just fails loudly if you point MONGO_URI at an older self-hosted server.\n", "version = server_version(coll)\n", "assert supports_rank_fusion(version), (\n", " f\"MongoDB {version} predates $rankFusion (needs 8.0+). Use an Atlas cluster (any tier) or a \"\n", " f\"self-hosted MongoDB 8.0+.\"\n", ")\n", "print(f\"server={version} hybrid search=$rankFusion (native)\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pattern 1: Vector search (`$vectorSearch`)\n", "\n", "Semantic recall: find documents whose `embedding` is nearest the query vector. This example\n", "reuses an existing document's embedding as the query (a pending case looking for decided\n", "precedent), so no embedding API call is needed. The builder filters to decided cases via the `status` field." ] }, { "cell_type": "code", "execution_count": 7, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "nearest decided precedents:\n", " txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third\n", " txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account\n", " txn-struct-03: Transfer of 4999 USD deliberately under 5000 to avoid reporting. Pattern\n" ] } ], "source": [ "query_doc = coll.find_one({\"transaction_id\": \"txn-review-struct\"}) # a pending case\n", "hits = list(coll.aggregate(build_vector_pipeline(query_doc[\"embedding\"], limit=3)))\n", "print(\"nearest decided precedents:\")\n", "for h in hits:\n", " print(f\" {h['transaction_id']}: {h['text'][:72]}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pattern 2: Full-text search (`$search`, BM25)\n", "\n", "Keyword and phrase matching over text fields — this is what catches the exact names, IDs, and\n", "codes that embeddings blur." ] }, { "cell_type": "code", "execution_count": 8, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "full-text matches:\n", " txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third\n", " txn-review-struct: Cash deposit of 4950 USD, just under the 5000 reporting threshold, follo\n", " txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account\n" ] } ], "source": [ "hits = list(\n", " coll.aggregate(\n", " build_lexical_pipeline(\"cash deposit just under the reporting threshold\", limit=3)\n", " )\n", ")\n", "print(\"full-text matches:\")\n", "for h in hits:\n", " print(f\" {h['transaction_id']}: {h['text'][:72]}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pattern 3: Hybrid (reciprocal rank fusion)\n", "\n", "Fuse the vector and full-text rankings into one result set. `$rankFusion` (MongoDB 8.0+) runs\n", "both input pipelines server-side and combines them by reciprocal rank — no client-side merging,\n", "no second round-trip." ] }, { "cell_type": "code", "execution_count": 9, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "hybrid via $rankFusion (server-side):\n", " txn-struct-01: Cash deposit of 4950 USD, just under the 5000 reporting threshold. Third\n", " txn-struct-02: Cash deposit of 4900 USD just below the 5000 CTR threshold, same account\n", " txn-struct-03: Transfer of 4999 USD deliberately under 5000 to avoid reporting. Pattern\n" ] } ], "source": [ "qvec, query = query_doc[\"embedding\"], \"structuring: cash deposit just under the threshold\"\n", "hits = list(coll.aggregate(build_rank_fusion_pipeline(qvec, query, k=3)))\n", "print(\"hybrid via $rankFusion (server-side):\")\n", "for h in hits:\n", " print(f\" {h['transaction_id']}: {h['text'][:72]}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Pattern 4: Graph traversal (`$graphLookup`)\n", "\n", "Follow `sender.account_number -> recipient.account_number` links to surface a network (here, a\n", "circular money-flow ring). This is a *relationship* signal, deliberately separate from the\n", "similarity ranking above." ] }, { "cell_type": "code", "execution_count": 10, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "graph traversal from ACC-RING-A: network_size=4 circular_flow=True suspicious=True\n" ] } ], "source": [ "graph_doc = next(\n", " iter(coll.aggregate(build_graph_pipeline(\"ACC-RING-A\", collection=coll.name))), {\"chain\": []}\n", ")\n", "ring = summarize_ring(graph_doc, seed_account=\"ACC-RING-A\")\n", "print(\n", " f\"graph traversal from ACC-RING-A: network_size={ring['network_size']} \"\n", " f\"circular_flow={ring['circular_flow']} suspicious={ring['suspicious_patterns']}\"\n", ")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Adapt these for your domain.** Each builder is a generic shape: embed your text with any\n", "model and point `build_vector_pipeline` at your field; set the searchable paths for full-text;\n", "tune the `$rankFusion` input weights to bias semantic vs. lexical; change the graph\n", "`connectFromField`/`connectToField` to your relationship (citations, org charts, supply chains).\n", "All four are defined in the cell above — lift them straight into your own project." ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 3. The end-to-end agent: human-in-the-loop fraud review\n", "\n", "Now everything works together. A Managed Agent reviews flagged transactions, grounding each\n", "recommendation in the MongoDB signals from Section 2 (exposed as **custom tools** over the\n", "Path A round-trip), and pauses for a human on the risky cases. Decisions and an append-only\n", "audit trail are written back to the same cluster.\n", "\n", "

\n", " \"The\n", "

\n", "\n", "The **agent loop runs on Anthropic**; the **data path runs in your notebook**. As a result,\n", "`MONGO_URI` and the embedding key never enter the agent or its sandbox. The human wait is durable — the\n", "session sits idle server-side via `requires_action` until you respond — while the simple\n", "streaming loop below is for development (drive it from a durable backend with the webhook\n", "pattern in production).\n", "\n", "### AP2 mandates: verify, then reason\n", "\n", "This agent reviews *agent-initiated* payments, so before any behavioral analysis it verifies an\n", "**AP2 (Agent Payments Protocol)** mandate — the cryptographically signed credential that proves a\n", "user authorized the payment. AP2 is an [open protocol](https://github.com/google-agentic-commerce/AP2)\n", "for agent-driven commerce. Its mandates (a **Checkout Mandate** and a **Payment Mandate**) are\n", "tamper-evident, signed digital objects forming an auditable chain. You do not need to learn the\n", "protocol to follow this cookbook: **call the `verify_mandates` tool and act on its verdict**\n", "(valid, constraints-satisfied, and double-spend). The signing and verification details — ES256 JWTs —\n", "stay encapsulated in [`ap2_mandates.py`](mongodb_on_cma/ap2_mandates.py), and the signed mandates and\n", "receipts are stored in MongoDB alongside the transactions.\n", "\n", "The cell below plays the \"Trusted Surface\" (generates a keypair, attaches signed mandates to the\n", "pending cases). The private key stays in local Python state — never the agent, a tool result, or\n", "the database." ] }, { "cell_type": "code", "execution_count": 11, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "attached AP2 mandates to 5 transactions agent_pk=04e1e32b7414d6f7…\n" ] } ], "source": [ "from cryptography.hazmat.primitives import serialization\n", "from cryptography.hazmat.primitives.asymmetric import ec\n", "\n", "ts_private_key = ec.generate_private_key(ec.SECP256R1()) # Trusted Surface key, fresh each run\n", "AGENT_PK = (\n", " ts_private_key.public_key()\n", " .public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)\n", " .hex()\n", ")\n", "ts_public_key = ts_private_key.public_key()\n", "\n", "PENDING = [\n", " \"txn-review-clean\",\n", " \"txn-review-fraud\",\n", " \"txn-review-struct\", # seeded human-override case under AUTO_APPROVE\n", " \"txn-review-high\",\n", " \"txn-review-ring\",\n", "]\n", "mandates = attach_mandates(coll, PENDING, agent_pk=AGENT_PK, ts_private_key=ts_private_key)\n", "print(f\"attached AP2 mandates to {len(mandates)} transactions agent_pk={AGENT_PK[:16]}…\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### The custom tools and host-side handlers\n", "\n", "Each tool is a `type: \"custom\"` tool. When the agent calls one, the session pauses, **this\n", "notebook** runs the handler with `pymongo`, and sends the JSON result back — the agent never\n", "touches the database or the connection string. The handler implementations are defined in the\n", "next cell: they wrap the Section 2 builders plus the AP2 verification (`verify_mandates`).\n", "`HANDLERS` maps only the five data tools; `escalate` is deliberately absent because the gate\n", "loop routes it to the human resolver instead of a handler (see the next section)." ] }, { "cell_type": "code", "execution_count": 12, "metadata": {}, "outputs": [], "source": [ "# The host-side tool implementations — the pymongo work behind each custom tool. These run in\n", "# THIS process (never the sandbox), reusing the Section 2 builders. The dispatch wiring + the\n", "# AP2/record-decision wrappers are in the next cell; build_decision_doc / build_audit_event\n", "# (imported above) shape the persisted documents and are shared with the AP2 module.\n", "\n", "\n", "def _jsonable(value):\n", " \"\"\"Recursively coerce Mongo/BSON values (ObjectId, datetime, ...) into JSON-safe types.\"\"\"\n", " if isinstance(value, dict):\n", " return {k: _jsonable(v) for k, v in value.items()}\n", " if isinstance(value, (list, tuple)):\n", " return [_jsonable(v) for v in value]\n", " if isinstance(value, datetime):\n", " return value.isoformat()\n", " if isinstance(value, (str, int, float, bool)) or value is None:\n", " return value\n", " return str(value)\n", "\n", "\n", "def _pick_fields(candidate):\n", " keep = (\"transaction_id\", \"text\", \"amount\", \"sender\", \"recipient\", \"decision\", \"score\")\n", " return _jsonable({k: candidate[k] for k in keep if k in candidate})\n", "\n", "\n", "def rerank_pool_size(k: int, fanout: int, *, enable_rerank: bool) -> int:\n", " return k * fanout if enable_rerank else k\n", "\n", "\n", "def merge_rerank(candidates, rerank_results, *, top_k) -> list[dict]:\n", " def _score(r):\n", " s = getattr(r, \"relevance_score\", None)\n", " return (\n", " (s if s is not None else r.get(\"relevance_score\", float(\"-inf\")))\n", " if isinstance(r, dict)\n", " else (s if s is not None else float(\"-inf\"))\n", " )\n", "\n", " seen: set[str] = set()\n", " out: list[dict] = []\n", " for result in sorted(rerank_results, key=_score, reverse=True):\n", " index = getattr(result, \"index\", None)\n", " if index is None and isinstance(result, dict):\n", " index = result.get(\"index\")\n", " score = getattr(result, \"relevance_score\", None)\n", " if score is None and isinstance(result, dict):\n", " score = result.get(\"relevance_score\")\n", " if index is None or index < 0 or index >= len(candidates):\n", " continue\n", " candidate = candidates[index]\n", " cid = str(candidate.get(\"transaction_id\", index))\n", " if cid in seen:\n", " continue\n", " seen.add(cid)\n", " out.append({**candidate, \"score\": score})\n", " if len(out) >= top_k:\n", " break\n", " return out\n", "\n", "\n", "def tool_get_transaction(coll, transaction_id) -> dict:\n", " doc = coll.find_one({\"transaction_id\": transaction_id})\n", " if not doc:\n", " return {\"error\": \"not_found\", \"transaction_id\": transaction_id}\n", " return _jsonable({k: v for k, v in doc.items() if k not in (\"embedding\", \"_id\")})\n", "\n", "\n", "def tool_hybrid_search_similar_frauds(\n", " coll,\n", " transaction_id,\n", " k,\n", " *,\n", " enable_rerank=False,\n", " fanout=5,\n", " reranker=None,\n", " vector_index=VECTOR_INDEX_NAME,\n", " search_index=SEARCH_INDEX_NAME,\n", " status_in=DECIDED_STATUSES,\n", ") -> dict:\n", " txn = coll.find_one({\"transaction_id\": transaction_id})\n", " if not txn:\n", " return {\"error\": \"not_found\", \"transaction_id\": transaction_id, \"similar\": []}\n", " qvec = txn[\"embedding\"]\n", " query = txn.get(\"text\", \"\")\n", " pool_k = rerank_pool_size(k, fanout, enable_rerank=enable_rerank)\n", " candidates = list(\n", " coll.aggregate(\n", " build_rank_fusion_pipeline(\n", " qvec,\n", " query,\n", " k=pool_k,\n", " vector_index=vector_index,\n", " search_index=search_index,\n", " status_in=status_in,\n", " )\n", " )\n", " )\n", " candidates = [c for c in candidates if str(c.get(\"transaction_id\")) != str(transaction_id)]\n", " if enable_rerank and reranker is not None and candidates:\n", " results = reranker(query, [c.get(\"text\", \"\") for c in candidates], top_k=k)\n", " candidates = merge_rerank(candidates, results, top_k=k)\n", " else:\n", " candidates = candidates[:k]\n", " return {\"similar\": [_pick_fields(c) for c in candidates]}\n", "\n", "\n", "def tool_detect_fraud_ring(coll, account_id, *, max_depth=4) -> dict:\n", " pipeline = build_graph_pipeline(account_id, max_depth=max_depth, collection=coll.name)\n", " docs = list(coll.aggregate(pipeline))\n", " return _jsonable(summarize_ring(docs[0] if docs else {\"chain\": []}, seed_account=account_id))\n", "\n", "\n", "def tool_record_decision(\n", " db,\n", " transaction_id,\n", " decision,\n", " *,\n", " confidence,\n", " risk_factors,\n", " reasoning,\n", " reviewed_by,\n", " escalated=False,\n", " recommended_decision=None,\n", ") -> dict:\n", " decision_doc = build_decision_doc(\n", " transaction_id,\n", " decision,\n", " confidence=confidence,\n", " risk_factors=risk_factors,\n", " reasoning=reasoning,\n", " reviewed_by=reviewed_by,\n", " )\n", " db[\"transaction_decisions\"].insert_one(decision_doc)\n", " if escalated:\n", " audit = build_audit_event(\n", " \"escalated_to_human\",\n", " transaction_id,\n", " decision_id=decision_doc[\"decision_id\"],\n", " severity=\"warning\",\n", " event_data={\"human_decision\": decision, \"recommended_decision\": recommended_decision},\n", " )\n", " else:\n", " audit = build_audit_event(\n", " \"decision_stored\", transaction_id, decision_id=decision_doc[\"decision_id\"]\n", " )\n", " db[\"audit_events\"].insert_one(audit)\n", " # Advance the lifecycle status to past tense (approve -> approved) so a decided case matches\n", " # the seed's vocabulary and DECIDED_STATUSES, making it eligible as precedent in later reviews.\n", " status = {\"approve\": \"approved\", \"reject\": \"rejected\"}.get(decision, decision)\n", " db[\"transactions\"].update_one({\"transaction_id\": transaction_id}, {\"$set\": {\"status\": status}})\n", " return {\"recorded\": True, \"decision_id\": decision_doc[\"decision_id\"]}" ] }, { "cell_type": "code", "execution_count": 13, "metadata": {}, "outputs": [], "source": [ "TOOLS = [\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"verify_mandates\",\n", " \"description\": \"Validate the AP2 Checkout and Payment Mandate JWTs (signature, constraints, \"\n", " \"double-spend). Run this FIRST. If valid=false, constraints_satisfied=false, or \"\n", " \"double_spend_detected=true, reject immediately.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\"transaction_id\": {\"type\": \"string\"}},\n", " \"required\": [\"transaction_id\"],\n", " },\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"get_transaction\",\n", " \"description\": \"Fetch the full transaction record under review.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\"transaction_id\": {\"type\": \"string\"}},\n", " \"required\": [\"transaction_id\"],\n", " },\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"hybrid_search_similar_frauds\",\n", " \"description\": \"Retrieve the most similar prior (already-decided) cases as precedent, using \"\n", " \"hybrid vector + full-text search.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\n", " \"transaction_id\": {\"type\": \"string\"},\n", " \"k\": {\"type\": \"integer\", \"description\": \"how many precedents (default 5)\"},\n", " },\n", " \"required\": [\"transaction_id\"],\n", " },\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"detect_fraud_ring\",\n", " \"description\": \"Trace the account's sender->recipient chain for circular-flow / mule / layering patterns.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\"account_id\": {\"type\": \"string\"}},\n", " \"required\": [\"account_id\"],\n", " },\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"record_decision\",\n", " \"description\": \"Persist the final approve/reject decision with reasoning and an audit event.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\n", " \"transaction_id\": {\"type\": \"string\"},\n", " \"decision\": {\"type\": \"string\", \"enum\": [\"approve\", \"reject\"]},\n", " \"confidence\": {\"type\": \"number\"},\n", " \"risk_factors\": {\"type\": \"array\", \"items\": {\"type\": \"string\"}},\n", " \"reasoning\": {\"type\": \"string\"},\n", " \"escalated\": {\"type\": \"boolean\"},\n", " \"recommended_decision\": {\"type\": \"string\", \"enum\": [\"approve\", \"reject\"]},\n", " },\n", " \"required\": [\"transaction_id\", \"decision\", \"reasoning\"],\n", " },\n", " },\n", " {\n", " \"type\": \"custom\",\n", " \"name\": \"escalate\",\n", " \"description\": \"Send a risky case to a human reviewer for the final decision. Use for \"\n", " \"medium-confidence, high-value, structuring, or fraud-ring cases.\",\n", " \"input_schema\": {\n", " \"type\": \"object\",\n", " \"properties\": {\n", " \"transaction_id\": {\"type\": \"string\"},\n", " \"recommended_decision\": {\"type\": \"string\", \"enum\": [\"approve\", \"reject\"]},\n", " \"confidence\": {\"type\": \"number\"},\n", " \"reason\": {\"type\": \"string\"},\n", " },\n", " \"required\": [\"transaction_id\", \"recommended_decision\", \"reason\"],\n", " },\n", " },\n", "]" ] }, { "cell_type": "code", "execution_count": 14, "metadata": {}, "outputs": [], "source": [ "# Host-side handlers — map each data tool to its handler function.\n", "reranker = (\n", " (lambda q, ds, top_k: rerank(q, ds, client=ai_client, top_k=top_k))\n", " if (ENABLE_RERANK and ai_client)\n", " else None\n", ")\n", "\n", "\n", "def _verify_mandates(inp):\n", " result = tool_verify_mandates(db, inp[\"transaction_id\"], ts_public_key)\n", " ok = result[\"valid\"] and result[\"constraints_satisfied\"] and not result[\"double_spend_detected\"]\n", " print(f\"\\n [verify_mandates] {inp['transaction_id']}: {'pass' if ok else 'FAIL'}\")\n", " return result\n", "\n", "\n", "def _hybrid(inp):\n", " return tool_hybrid_search_similar_frauds(\n", " coll,\n", " inp[\"transaction_id\"],\n", " inp.get(\"k\", 5),\n", " enable_rerank=ENABLE_RERANK,\n", " reranker=reranker,\n", " )\n", "\n", "\n", "def _record(inp):\n", " result = tool_record_decision(\n", " db,\n", " inp[\"transaction_id\"],\n", " inp[\"decision\"],\n", " confidence=inp.get(\"confidence\", 0),\n", " risk_factors=inp.get(\"risk_factors\", []),\n", " reasoning=inp.get(\"reasoning\", \"\"),\n", " reviewed_by=\"human\" if inp.get(\"escalated\") else \"agent\",\n", " escalated=inp.get(\"escalated\", False),\n", " recommended_decision=inp.get(\"recommended_decision\"),\n", " )\n", " if (\n", " inp[\"decision\"] == \"approve\"\n", " ): # store the AP2 receipt that later powers double-spend detection\n", " txn_doc = coll.find_one(\n", " {\"transaction_id\": inp[\"transaction_id\"]},\n", " {\"checkout_mandate_jwt\": 1, \"mandate_id\": 1, \"agent_pk\": 1},\n", " )\n", " if txn_doc and txn_doc.get(\"mandate_id\"):\n", " checkout_hash = hashlib.sha256(txn_doc[\"checkout_mandate_jwt\"].encode()).hexdigest()\n", " store_mandate_receipt(\n", " db, txn_doc[\"mandate_id\"], txn_doc[\"agent_pk\"], checkout_hash, \"approve\"\n", " )\n", " return result\n", "\n", "\n", "HANDLERS = {\n", " \"verify_mandates\": _verify_mandates,\n", " \"get_transaction\": lambda inp: tool_get_transaction(coll, inp[\"transaction_id\"]),\n", " \"hybrid_search_similar_frauds\": _hybrid,\n", " \"detect_fraud_ring\": lambda inp: tool_detect_fraud_ring(coll, inp[\"account_id\"]),\n", " \"record_decision\": _record,\n", "}" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Create the agent, environment, and session\n", "\n", "`model`, `system`, and `tools` live on the **agent**. The session references it and provisions\n", "the sandbox. Networking is `limited`: the agent reaches MongoDB only through the host-side round-trip." ] }, { "cell_type": "code", "execution_count": 15, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "session: sesn_01LbAUC1V36CQS3Sk26eixTc\n", "Watch in Console: https://platform.claude.com/workspaces/default/sessions/sesn_01LbAUC1V36CQS3Sk26eixTc\n" ] } ], "source": [ "SYSTEM = \"\"\"You are a financial fraud reviewer. You will be given transaction IDs to review.\n", "\n", "For EACH transaction, in order:\n", "1. Call get_transaction to read it.\n", "2. Call verify_mandates to validate the AP2 mandate chain (signature, constraints, double-spend).\n", " HARD GATE: if verify_mandates returns valid=false, constraints_satisfied=false, OR\n", " double_spend_detected=true, call record_decision immediately with decision=\"reject\" and\n", " skip the remaining steps for that transaction.\n", "3. Call hybrid_search_similar_frauds to retrieve similar decided precedents.\n", "4. Call detect_fraud_ring on the sender's account_number to check for ring/mule patterns.\n", "5. Weigh the precedents, the ring signal, the amount, and your confidence. Then make EXACTLY\n", " ONE terminal call for that transaction:\n", " - You MUST call escalate (do NOT call record_decision yourself) whenever ANY of these holds:\n", " * a structuring amount ($4,900-$4,999),\n", " * a high-value amount (>= $50,000) that you would otherwise approve,\n", " * detect_fraud_ring reports suspicious_patterns, or\n", " * your confidence is medium (~75-85).\n", " Give your recommended_decision and a short reason.\n", " - Otherwise (a clear-cut case matching none of the above), call record_decision (approve/reject).\n", "\n", "When you escalate, you will receive the human's decision. Then call record_decision with that\n", "decision, escalated=true, and recommended_decision set to what you had recommended.\n", "\n", "Be concise. Move to the next transaction after recording a decision.\"\"\"\n", "\n", "agent = client.beta.agents.create(\n", " name=\"MongoDB Atlas fraud reviewer\", model=MODEL, system=SYSTEM, tools=TOOLS\n", ")\n", "environment = client.beta.environments.create(\n", " name=\"fraud-review-env\", config={\"type\": \"cloud\", \"networking\": {\"type\": \"limited\"}}\n", ")\n", "session = client.beta.sessions.create(\n", " agent={\"type\": \"agent\", \"id\": agent.id, \"version\": agent.version},\n", " environment_id=environment.id,\n", " title=\"Fraud review\",\n", ")\n", "print(\"session:\", session.id)\n", "print(f\"Watch in Console: https://platform.claude.com/workspaces/default/sessions/{session.id}\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Run the review with the human gate\n", "\n", "Open the event stream, send the queue of pending transactions, and drive the gate loop with\n", "`run_gate_loop`: data-tool calls are serviced automatically from `HANDLERS`, and an `escalate`\n", "call pauses for a human. With `AUTO_APPROVE` set (for example, in CI), the gate resolves\n", "deterministically and **overrides the agent on the seeded structuring case** so a differing\n", "human verdict is visible. Otherwise, you are prompted inline." ] }, { "cell_type": "code", "execution_count": 16, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "\n", " [verify_mandates] txn-review-clean: pass\n", "\n", " [verify_mandates] txn-review-fraud: pass\n", "\n", " [verify_mandates] txn-review-struct: pass\n", "\n", " [verify_mandates] txn-review-high: pass\n", "\n", " [verify_mandates] txn-review-ring: pass\n", "\n", " 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.\n" ] }, { "name": "stdin", "output_type": "stream", "text": [ " Your decision [approve/reject]: reject\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\n", " 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.\n" ] }, { "name": "stdin", "output_type": "stream", "text": [ " Your decision [approve/reject]: approve\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\n", " 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.\n" ] }, { "name": "stdin", "output_type": "stream", "text": [ " Your decision [approve/reject]: reject\n" ] }, { "name": "stdout", "output_type": "stream", "text": [ "\n", "serviced 28 tool calls\n" ] } ], "source": [ "OVERRIDE_IDS = {\"txn-review-struct\"}\n", "\n", "\n", "def resolve_human_decision(recommendation, *, auto_approve, override_ids, txn_id) -> str:\n", " \"\"\"Stand in for a human reviewer. In CI (auto_approve) it concurs with the agent, except on\n", " the seeded override_ids where it flips the verdict so a differing human decision is visible.\"\"\"\n", " opposite = \"reject\" if recommendation == \"approve\" else \"approve\"\n", " if auto_approve and txn_id in set(override_ids):\n", " return opposite\n", " return recommendation\n", "\n", "\n", "def resolver(tool_input):\n", " txn_id = tool_input.get(\"transaction_id\", \"\")\n", " recommended = tool_input.get(\"recommended_decision\", \"reject\")\n", " if AUTO_APPROVE:\n", " decision = resolve_human_decision(\n", " recommended, auto_approve=True, override_ids=OVERRIDE_IDS, txn_id=txn_id\n", " )\n", " flag = \" <-- HUMAN OVERRIDE\" if decision != recommended else \"\"\n", " print(f\" [human/auto] {txn_id}: agent recommended {recommended} -> human {decision}{flag}\")\n", " return decision\n", " print(\n", " f\"\\n ESCALATED {txn_id} (agent recommends {recommended}): {tool_input.get('reason', '')}\"\n", " )\n", " return (\n", " \"approve\"\n", " if input(\" Your decision [approve/reject]: \").strip().lower().startswith(\"a\")\n", " else \"reject\"\n", " )\n", "\n", "\n", "def send_result(custom_tool_use_id, result):\n", " client.beta.sessions.events.send(\n", " session_id=session.id,\n", " events=[\n", " {\n", " \"type\": \"user.custom_tool_result\",\n", " \"custom_tool_use_id\": custom_tool_use_id,\n", " \"content\": [{\"type\": \"text\", \"text\": json.dumps(result)}],\n", " }\n", " ],\n", " )\n", "\n", "\n", "def run_gate_loop(stream, send, *, handlers, resolver) -> dict:\n", " \"\"\"Drive the session to completion, servicing custom-tool calls as they arrive.\n", "\n", " This is the whole Path A round-trip for a real workload. Each `agent.custom_tool_use`\n", " event names a tool and its input; the session then goes idle with\n", " `stop_reason == \"requires_action\"` until we answer. We look up the tool in `handlers`\n", " (which run `pymongo` host-side — the credential never leaves this process), and route the\n", " special `escalate` tool to the human `resolver` instead. We post each result back with\n", " `send`, and the session resumes. Break on a terminal idle (`end_turn`) or termination.\n", " \"\"\"\n", " pending: dict[str, Any] = {}\n", " responded: set[str] = set()\n", " serviced: list[tuple[str, dict]] = []\n", " for ev in stream:\n", " if ev.type == \"agent.custom_tool_use\":\n", " pending[ev.id] = ev\n", " elif ev.type == \"session.status_idle\":\n", " stop = getattr(ev, \"stop_reason\", None)\n", " if stop is None:\n", " continue\n", " if stop.type != \"requires_action\":\n", " break # end_turn / retries_exhausted — the run is done\n", " for event_id in stop.event_ids or []:\n", " if event_id in responded:\n", " continue\n", " call = pending.get(event_id)\n", " if call is None:\n", " continue\n", " if call.name == \"escalate\":\n", " result = {\"human_decision\": resolver(call.input)}\n", " else:\n", " handler = handlers.get(call.name)\n", " result = (\n", " handler(call.input) if handler else {\"error\": f\"unknown tool {call.name}\"}\n", " )\n", " responded.add(event_id)\n", " send(event_id, result)\n", " serviced.append((call.name, result))\n", " elif ev.type == \"session.status_terminated\":\n", " break\n", " return {\"serviced\": serviced, \"responded\": sorted(responded)}\n", "\n", "\n", "kickoff = {\n", " \"type\": \"user.message\",\n", " \"content\": [\n", " {\"type\": \"text\", \"text\": \"Review these flagged transactions: \" + \", \".join(PENDING)}\n", " ],\n", "}\n", "run_start = datetime.now(UTC)\n", "\n", "with client.beta.sessions.events.stream(session_id=session.id) as stream:\n", " client.beta.sessions.events.send(session_id=session.id, events=[kickoff])\n", " gate_result = run_gate_loop(stream, send_result, handlers=HANDLERS, resolver=resolver)\n", "\n", "wait_for_idle_status(client, session.id)\n", "print(f\"\\nserviced {len(gate_result['serviced'])} tool calls\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "### Review the decisions and audit trail\n", "\n", "Read the decisions and audit trail back from MongoDB. The decision and audit collections are\n", "**append-only**, so the read-back is scoped to this run. The closing asserts make this a\n", "**self-check**: a re-run that doesn't reproduce the expected lane mix fails loudly.\n", "\n", "The human-override count reflects how the gate was resolved: with `AUTO_APPROVE` it is 1 (the\n", "seeded override on `txn-review-struct` is shown), while in an interactive run it is 0 whenever\n", "you concur with the agent's recommendation on every escalation — both are expected." ] }, { "cell_type": "code", "execution_count": 17, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "Decisions by lane:\n", " clean_approve -> approve: 1\n", " clear_reject -> reject: 1\n", " high_value -> approve: 1\n", " ring -> reject: 1\n", " structuring -> reject: 1\n", "\n", "escalated_to_human events: 3\n", "human overrides (verdict != agent recommendation): 0\n", "\n", "self-check OK\n" ] } ], "source": [ "from collections import Counter\n", "\n", "decisions = list(db[\"transaction_decisions\"].find({\"created_at\": {\"$gte\": run_start}}, {\"_id\": 0}))\n", "audits = list(db[\"audit_events\"].find({\"timestamp\": {\"$gte\": run_start}}, {\"_id\": 0}))\n", "\n", "print(\"Decisions by lane:\")\n", "lanes = Counter()\n", "for d in decisions:\n", " txn = coll.find_one({\"transaction_id\": d[\"transaction_id\"]})\n", " lanes[f\"{(txn['lane'] if txn else '?')} -> {d['decision']}\"] += 1\n", "for k, v in sorted(lanes.items()):\n", " print(f\" {k}: {v}\")\n", "\n", "overrides = [\n", " a\n", " for a in audits\n", " if a[\"event_type\"] == \"escalated_to_human\"\n", " and a[\"event_data\"].get(\"human_decision\") != a[\"event_data\"].get(\"recommended_decision\")\n", "]\n", "print(\n", " f\"\\nescalated_to_human events: {sum(1 for a in audits if a['event_type'] == 'escalated_to_human')}\"\n", ")\n", "print(f\"human overrides (verdict != agent recommendation): {len(overrides)}\")\n", "for a in overrides:\n", " ed = a[\"event_data\"]\n", " print(\n", " f\" {a['transaction_id']}: agent {ed['recommended_decision']} -> human {ed['human_decision']}\"\n", " )\n", "\n", "# Self-check — fail loudly if this run didn't produce the expected result.\n", "decided_lanes = {coll.find_one({\"transaction_id\": d[\"transaction_id\"]})[\"lane\"] for d in decisions}\n", "assert decided_lanes == {\"clean_approve\", \"clear_reject\", \"high_value\", \"ring\", \"structuring\"}, (\n", " f\"expected all 5 lanes decided, saw {sorted(decided_lanes)}\"\n", ")\n", "assert len(decisions) == len(PENDING), f\"expected {len(PENDING)} decisions, got {len(decisions)}\"\n", "assert sum(1 for a in audits if a[\"event_type\"] == \"escalated_to_human\") >= 1, (\n", " \"expected >=1 escalation\"\n", ")\n", "if AUTO_APPROVE:\n", " assert overrides, \"AUTO_APPROVE run should show the seeded override on txn-review-struct\"\n", "print(\"\\nself-check OK\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 4. MongoDB Atlas as the system of record and audit backbone\n", "\n", "Step back and look at where the run's state lives: everything the agent did is durable in the\n", "same cluster it retrieved from. That is the single-engine payoff on the write side —\n", "`transactions` is the operational record (each case advances `pending` → `approved` or\n", "`rejected`), `transaction_decisions` holds immutable verdicts, `audit_events` is an append-only\n", "trail, and `mandate_receipts` powers double-spend detection. One case, end to end:" ] }, { "cell_type": "code", "execution_count": 18, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "operational record: txn-review-struct lane=structuring status=rejected\n", "decision record: reject reviewed_by=human confidence=0.9\n", "audit trail (append-only):\n", " 13:49:22 escalated_to_human severity=warning (agent recommended reject, human decided reject)\n" ] } ], "source": [ "case = \"txn-review-struct\"\n", "txn = coll.find_one({\"transaction_id\": case}, {\"_id\": 0, \"embedding\": 0})\n", "print(f\"operational record: {case} lane={txn['lane']} status={txn['status']}\")\n", "\n", "decision = db[\"transaction_decisions\"].find_one(\n", " {\"transaction_id\": case, \"created_at\": {\"$gte\": run_start}}, {\"_id\": 0}\n", ")\n", "print(\n", " f\"decision record: {decision['decision']} reviewed_by={decision['reviewed_by']} confidence={decision['confidence_score']}\"\n", ")\n", "\n", "print(\"audit trail (append-only):\")\n", "for a in (\n", " db[\"audit_events\"]\n", " .find({\"transaction_id\": case, \"timestamp\": {\"$gte\": run_start}}, {\"_id\": 0})\n", " .sort(\"timestamp\", 1)\n", "):\n", " extra = \"\"\n", " if a[\"event_type\"] == \"escalated_to_human\":\n", " ed = a[\"event_data\"]\n", " extra = f\" (agent recommended {ed['recommended_decision']}, human decided {ed['human_decision']})\"\n", " print(f\" {a['timestamp']:%H:%M:%S} {a['event_type']} severity={a['severity']}{extra}\")\n", "\n", "txn_mandate = coll.find_one({\"transaction_id\": case}, {\"mandate_id\": 1, \"agent_pk\": 1})\n", "if txn_mandate and txn_mandate.get(\"mandate_id\"):\n", " receipt = db[\"mandate_receipts\"].find_one({\"mandate_id\": txn_mandate[\"mandate_id\"]}, {\"_id\": 0})\n", " if receipt:\n", " print(\n", " f\"mandate receipt: mandate_id={receipt['mandate_id'][:12]}… decision={receipt['decision']}\"\n", " )" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "**Webhooks for production (pointer, not run here):** The streaming loop above is great for\n", "development but does not survive a restart. For production, register a webhook for\n", "`session.status_idled`: your handler queues the case for a reviewer and later POSTs the\n", "`user.custom_tool_result`. The **data-tool handlers are identical**; only the trigger changes.\n", "See [`CMA_operate_in_production.ipynb`](CMA_operate_in_production.ipynb)." ] }, { "cell_type": "code", "execution_count": 19, "metadata": {}, "outputs": [ { "name": "stdout", "output_type": "stream", "text": [ "archived session + environment; closed MongoDB connection\n" ] } ], "source": [ "client.beta.sessions.archive(session.id)\n", "client.beta.environments.archive(environment.id)\n", "# Agents are reusable across runs; archiving is optional and permanent.\n", "mongo.close()\n", "print(\"archived session + environment; closed MongoDB connection\")" ] }, { "cell_type": "markdown", "metadata": {}, "source": [ "## 5. Recap\n", "\n", "You connected MongoDB Atlas to a Claude Managed Agent and took it from connection to system of\n", "record in one build:\n", "\n", "- **Connect** — three credential-safe data paths (host-side custom tool, self-hosted sandbox,\n", " self-hosted MCP), the MongoDB secret always on your side of the boundary.\n", "- **Retrieve** — vector, full-text, hybrid RRF, and graph traversal as liftable pipeline\n", " builders, defined inline in Section 2.\n", "- **Gate** — risky decisions behind CMA's native `requires_action` human pause.\n", "- **Persist** — decisions, an append-only audit trail, and AP2 receipts in the same cluster the\n", " agent retrieves from.\n", "\n", "The integration surface was small: a few lines of `pymongo` behind a custom tool. Swap the\n", "collection and the tools, and the same shape grounds any agent in MongoDB. The retrieval\n", "builders, tool handlers, and gate loop are all right here in the notebook; the\n", "[`mongodb_on_cma/`](mongodb_on_cma/) package holds only the setup boilerplate." ] } ], "metadata": { "kernelspec": { "display_name": "Python 3 (ipykernel)", "language": "python", "name": "python3" }, "language_info": { "codemirror_mode": { "name": "ipython", "version": 3 }, "file_extension": ".py", "mimetype": "text/x-python", "name": "python", "nbconvert_exporter": "python", "pygments_lexer": "ipython3", "version": "3.11.13" } }, "nbformat": 4, "nbformat_minor": 4 }