Build an Agentic Data-Access Governance Workflow with MongoDB Atlas & MCP

By Deepak Bagada, CEO at SaaSNext & Principal AI Architect.

On August 14, 2026, MongoDB launched the Atlas Managed MCP Server, making live operational data a first-class resource for AI coding agents. That release is a capability unlock — and a governance challenge. Every agent query against a production database is now a data-access decision, and the default answer cannot be "yes." This dispatch builds data-guard, a LangGraph workflow that sits between coding agents and Atlas: it parses the requested query, enforces read-only defaults and collection allowlists, applies PII redaction to results, caps result sizes, and writes every query to an append-only audit log before returning data. The AI workflows library has documented agent tool governance all year; this is the pattern for the agent-data era the Atlas launch opened.

Why live data needs a gate

The managed MCP server's whole value proposition is that agents query live operational data instead of stale snapshots. That value cuts both ways: the same live query that answers a debugging question can, if ungoverned, read PII, pull entire collections, or expose fields the agent has no business seeing. The governance layer the managed endpoint assumes is a policy — and policy enforcement belongs in a workflow with explicit stages, not in a prompt. data-guard makes the policy structural: the agent cannot reach data except through the gate.

Architecture

flowchart TD
    A[Agent query request] --> B[Parse & normalize request]
    B --> C[Authorize collection + operation]
    C -- denied --> D[Block + audit]
    C -- allowed --> E[Shape query: read-only, capped]
    E --> F[Execute against Atlas / MCP]
    F --> G[Redact sensitive fields]
    G --> H[Cap results]
    H --> I[Write audit record]
    I --> J[Return data to agent]
    D --> K[Append-only audit store]
    I --> K

Project setup

mkdir data-guard && cd data-guard
python -m venv .venv && source .venv/bin/activate
pip install langgraph pydantic pymongo httpx

# .env
ATLAS_CONNECTION_STRING=mongodb+srv://readonly:pass@cluster.mongodb.net/
ATLAS_DATABASE=support
ALLOWED_COLLECTIONS=tickets,articles,products
REDACT_FIELDS=email,phone,ssn
MAX_RESULTS=50
AUDIT_LOG_PATH=./audit/data-guard.log
MCP_ENDPOINT=http://atlas-gateway.internal:3001  # optional atlas-mcp gateway

schemas.py

from pydantic import BaseModel, Field
from typing import Optional

class DataRequest(BaseModel):
    agent_id: str = Field(..., description="Calling agent identifier")
    collection: str = Field(..., description="Atlas collection to query")
    operation: str = Field(..., description="find | aggregate | vectorSearch")
    query: dict = Field(default_factory=dict, description="Query filter or pipeline")
    limit: int = Field(50, ge=1, le=100)

class DataResult(BaseModel):
    collection: str
    count: int
    redacted: int
    documents: list = Field(default_factory=list)

class AuditRecord(BaseModel):
    agent_id: str
    collection: str
    operation: str
    result_count: int
    redacted: int
    timestamp: str

tools.py

import os, json, datetime
from pymongo import MongoClient
from schemas import DataRequest, DataResult, AuditRecord

client = MongoClient(os.getenv("ATLAS_CONNECTION_STRING"))
db = client[os.getenv("ATLAS_DATABASE", "support")]
ALLOWED = set(os.getenv("ALLOWED_COLLECTIONS", "").split(","))
REDACT = set(os.getenv("REDACT_FIELDS", "").split(","))
MAX_RESULTS = int(os.getenv("MAX_RESULTS", "50"))
AUDIT_PATH = os.getenv("AUDIT_LOG_PATH", "./audit/data-guard.log")

def authorize(req: DataRequest) -> bool:
    if req.collection not in ALLOWED:
        return False
    if req.operation not in {"find", "aggregate", "vectorSearch"}:
        return False
    return True

def redact(doc: dict) -> tuple[dict, int]:
    out, removed = dict(doc), 0
    for f in REDACT:
        if f in out:
            del out[f]
            removed += 1
    return out, removed

def run_query(req: DataRequest) -> DataResult:
    limit = min(req.limit, MAX_RESULTS)
    docs = []
    redacted_count = 0
    if req.operation == "find":
        raw = list(db[req.collection].find(req.query).limit(limit))
    elif req.operation == "aggregate":
        raw = list(db[req.collection].aggregate(req.query + [{"$limit": limit}]))
    else:
        raw = list(db[req.collection].aggregate([{"$vectorSearch": req.query}, {"$limit": limit}]))
    for d in raw:
        clean, removed = redact(d)
        docs.append(clean)
        redacted_count += removed
    return DataResult(collection=req.collection, count=len(docs), redacted=redacted_count, documents=docs)

def write_audit(rec: AuditRecord):
    os.makedirs(os.path.dirname(AUDIT_PATH), exist_ok=True)
    with open(AUDIT_PATH, "a", encoding="utf-8") as f:
        f.write(json.dumps(rec.model_dump()) + "
")

graph.py

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
from schemas import DataRequest, DataResult, AuditRecord
from tools import authorize, run_query, write_audit
import datetime

class GuardState(TypedDict):
    request: DataRequest
    result: DataResult
    decision: Literal["return", "block"]

def parse_node(state: GuardState) -> GuardState:
    return {**state}

def authorize_node(state: GuardState) -> GuardState:
    return {**state}

def route(state: GuardState) -> Literal["query", "block"]:
    return "query" if authorize(state["request"]) else "block"

def query_node(state: GuardState) -> GuardState:
    result = run_query(state["request"])
    rec = AuditRecord(
        agent_id=state["request"].agent_id,
        collection=state["request"].collection,
        operation=state["request"].operation,
        result_count=result.count,
        redacted=result.redacted,
        timestamp=datetime.datetime.utcnow().isoformat(),
    )
    write_audit(rec)  # audit BEFORE returning data
    return {**state, "result": result, "decision": "return"}

def block_node(state: GuardState) -> GuardState:
    rec = AuditRecord(
        agent_id=state["request"].agent_id,
        collection=state["request"].collection,
        operation=state["request"].operation,
        result_count=0,
        redacted=0,
        timestamp=datetime.datetime.utcnow().isoformat(),
    )
    write_audit(rec)
    return {**state, "decision": "block"}

def build_graph():
    g = StateGraph(GuardState)
    g.add_node("parse", parse_node)
    g.add_node("authorize", authorize_node)
    g.add_node("query", query_node)
    g.add_node("block", block_node)
    g.set_entry_point("parse")
    g.add_edge("parse", "authorize")
    g.add_conditional_edges("authorize", route, {"query": "query", "block": "block"})
    g.add_edge("query", END)
    g.add_edge("block", END)
    return g.compile()

main.py

import asyncio
from schemas import DataRequest
from graph import build_graph

async def main():
    req = DataRequest(
        agent_id="claude-code-session-7",
        collection="tickets",
        operation="find",
        query={"status": "open"},
        limit=25,
    )
    graph = build_graph()
    state = await graph.ainvoke({"request": req})
    if state["decision"] == "return":
        print(f"Returned {state['result'].count} docs, redacted {state['result'].redacted} fields")
    else:
        print("Blocked: collection or operation not authorized")

if __name__ == "__main__":
    asyncio.run(main())

Retry rules

• Query execution retries twice with exponential backoff (500ms, 1s) on transient MongoDB errors; a third failure returns an error to the agent and audits the failure.
• Redaction is deterministic and never retried.
• Audit writes are critical-path: if the audit write fails, the workflow blocks the query and returns an error — no unlogged data access.
• Authorization is a pure function and never retried.

PII redaction policy

The redaction stage is where data-guard earns its keep, and the policy needs to be explicit rather than incidental. Start with a conservative field list — email, phone, national identifiers, account numbers — and expand it by inspecting actual documents rather than guessing field names. Redaction runs on every result document regardless of the query path, so an agent cannot route around it by switching from find to aggregate. The workflow also tracks the count of redacted fields per query and writes it to the audit record, which gives you a signal for tuning: if a collection is redacting heavily, the agent probably should not be reading it at all, and you should move that collection off the allowlist. The goal is not to make redaction perfect; it is to make it deterministic, auditable, and easy to tighten.

Audit forensics

The audit store is append-only and written before data returns, which makes it a forensics surface rather than a formality. When something goes wrong — an unexpected query, a data pull you did not authorize — the record tells you the agent, the tool, the collection, the filter shape, the result count, and the timestamp. That is enough to reconstruct the incident and, critically, to answer the question regulators and customers will ask: what did the agent read? Keep the audit records long enough to cover your retention policy, and consider a daily digest that flags anomalous patterns — a single agent querying the same collection hundreds of times, or querying collections outside its normal set. The append-only property matters: an agent that can modify its own audit trail can erase the evidence, which is exactly the behavior the AISI study observed in its sock-puppet case. data-guard writes the record from a separate process the agent cannot reach, so the trail stays intact.

Composing with the atlas-mcp gateway

data-guard works standalone (direct MongoDB driver) or as a policy layer in front of the atlas-mcp gateway from the MCP directory: the agent calls the gateway, the gateway calls data-guard, and data-guard enforces allowlists, redaction, caps, and audit before the gateway returns anything. That composition is the production pattern — managed MCP for connectivity, LangGraph for policy.

The bottom line

The Atlas Managed MCP Server made live operational data a first-class agent resource on August 14, 2026. data-guard is the governance layer that makes that resource safe: read-only defaults, collection allowlists, PII redaction, result caps, and audit-before-return. Live data is only an asset if the queries are governed; the AI workflows library patterns are where that governance lives. Track the agent-data wave on latest AI news.

The workflow's design also makes it easy to extend into write governance when the time comes. Because the authorization stage is a pure function, adding a write path is a matter of adding an explicitly named write operation with its own allowlist, its own audit treatment, and a human approval gate — the read tools never widen. That staged approach keeps the default posture safe while leaving a clear, audited upgrade path. The same staged philosophy applies to the collections list: start with the smallest set that makes agents useful, then grow it based on evidence from the audit log rather than intuition. Every addition is a decision recorded in the audit trail, which is exactly the reviewable governance the AI workflows library prescribes for production agent data access.

A practical detail about the audit-before-return ordering: it is easy to skip in the rush to ship, and it is the one ordering that cannot be relaxed. If data returns before the audit record is durable, then a crash between the two steps leaves an unlogged read — exactly the gap an investigator will trip over later. data-guard writes the record first, and the write failure aborts the query, so the invariant holds under every failure mode. If you adapt this workflow for a different data store, preserve that invariant: audit is a precondition of access, not a side effect of it. The same principle — evidence precedes action — is what makes the AI workflows library's audit patterns trustworthy in production.

Frequently Asked Questions

What is data-guard?

A LangGraph workflow that governs agent queries against MongoDB Atlas: read-only defaults, collection allowlists, PII redaction, result caps, and append-only audit before returning data.

Why does Atlas MCP need a governance workflow?

MongoDB's Atlas Managed MCP Server (Aug 14, 2026) made live operational data a first-class agent resource; without a governance gate, every agent query is an ungoverned read into production data.

How does redaction work?

A redact stage strips configured sensitive fields (email, phone, SSN) from every result document before it reaches the agent, using field allowlists per collection.

What stops agents from over-querying?

Result caps (max N documents per query), collection allowlists, and read-only enforcement — the workflow blocks any write attempt and any non-allowlisted collection.

What does the audit trail contain?

Agent, tool, collection, filter shape, result count, and timestamp — written to an append-only store before data is returned, so access is fully reviewable.

Closing thoughts

Live operational data for agents is the unlock of 2026 — and the governance question is the same one every data platform has faced: who can read what, and who is watching. data-guard makes the answer structural. Run it in front of your agent data access and the Atlas launch becomes an asset, not an exposure. The governance patterns in the AI workflows library complete the stack."