Audit Logging, Rate Limiting & Anomaly Detection

Prevention (Lessons 19-20) reduces the chance of a successful attack; detection and containment are what catch what prevention misses, and what make an eventual incident investigable. Every enterprise MCP server needs both, and needs to treat the audit log itself as a system requiring its own access controls, since it can accumulate exactly the sensitive data it exists to help protect.

Structured Audit Logging

Log enough on every tool call to answer “who did what, when, with what result” without needing to reconstruct context from scattered application logs.

import time, uuid, logging

audit_logger = logging.getLogger("mcp.audit")

SENSITIVE_ARG_PATTERNS = {"password", "ssn", "card_number", "api_key", "secret"}

def _redact_sensitive(arguments: dict) -> dict:
    """Redact known-sensitive field names before they ever reach the log,
    since audit logs otherwise inherit whatever sensitive data a caller
    happens to pass, verbatim, as a tool argument."""
    return {
        k: ("[REDACTED]" if any(p in k.lower() for p in SENSITIVE_ARG_PATTERNS) else v)
        for k, v in arguments.items()
    }

@mcp.middleware()
async def audit_log_middleware(request, call_next):
    call_id = str(uuid.uuid4())
    start = time.monotonic()

    audit_entry = {
        "call_id": call_id,
        "tool": request.tool_name,
        "subject": getattr(request.state, "subject", "unknown"),
        "session_id": request.session_id,
        "arguments": _redact_sensitive(request.arguments),
        "timestamp": time.time(),
    }

    try:
        result = await call_next(request)
        audit_entry["outcome"] = "success"
        audit_entry["duration_ms"] = (time.monotonic() - start) * 1000
        return result
    except Exception as e:
        audit_entry["outcome"] = "error"
        audit_entry["error"] = str(e)
        raise
    finally:
        audit_logger.info(audit_entry)  # Shipped to centralized log storage, Module 9

Every high-impact tool (writes, exports, deletions) should log at this granularity as a baseline; read-only, low-risk tools can log at a coarser sampling rate if call volume makes full logging impractical, but the writes always get full fidelity. The redaction step above matters precisely because field-name-based redaction only catches sensitive data in fields whose name signals sensitivity, a customer accidentally pasting a card number into a free-text description field (as in the quiz scenario) would not be caught by name-based redaction, which is a real limitation worth knowing rather than assuming the redaction list solves the problem completely; content-based scanning (regex for card-number-shaped strings, for instance) is a further layer worth adding for genuinely high-risk free-text fields.

Audit Logs Need Their Own Access Control and Retention Policy

Because the audit log captures call arguments verbatim (redaction gaps notwithstanding), it can become, over time, a repository containing exactly the kind of sensitive data the rest of this course’s authorization controls (Module 4) are designed to restrict access to. Treating “it’s just a log” as exempt from that same discipline is a common mistake. The log storage itself needs access control scoped to those who genuinely need to investigate incidents (typically a security or platform on-call rotation, not every engineer with general log access), and a defined retention period after which entries are purged, rather than accumulating indefinitely and growing the exposure surface with no corresponding benefit past whatever window is actually useful for investigation and compliance.

# Retention: purge audit entries past the org's defined window (e.g. 90 days),
# balancing investigation usefulness against unbounded accumulation of sensitive data
async def purge_old_audit_logs(retention_days: int = 90):
    cutoff = datetime.utcnow() - timedelta(days=retention_days)
    await audit_store.delete_older_than(cutoff)

Per-Caller Rate Limiting

Distinct from respecting a downstream provider’s rate limit (Lesson 15), this control limits how much any single caller can do against your own server, catching a single runaway or compromised agent rather than protecting a third-party API.

from collections import defaultdict
import time

class PerCallerRateLimiter:
    def __init__(self, max_calls: int, window_seconds: int):
        self.max_calls = max_calls
        self.window = window_seconds
        self._calls: dict[str, list[float]] = defaultdict(list)

    def check(self, caller_id: str) -> bool:
        now = time.time()
        recent = [t for t in self._calls[caller_id] if now - t < self.window]
        self._calls[caller_id] = recent
        if len(recent) >= self.max_calls:
            return False
        recent.append(now)
        return True

caller_limiter = PerCallerRateLimiter(max_calls=100, window_seconds=60)

@mcp.middleware()
async def rate_limit_middleware(request, call_next):
    caller_id = request.state.subject
    if not caller_limiter.check(caller_id):
        return {"isError": True, "message": "Rate limit exceeded, slow down and retry shortly"}
    return await call_next(request)

Anomaly Signals Specific to Agent Tool Calls

flowchart TD
    Logs["Audit log stream"] --> Detect["Anomaly detection rules"]
    Detect --> S1["Sudden spike in a high-impact\ntool's call volume from one caller"]
    Detect --> S2["A caller invoking tools outside\nits normal historical pattern"]
    Detect --> S3["A burst of authorization failures\n(403s) from one caller,\nsuggests scope-probing"]
    Detect --> S4["Export/read-all tools called\noutside business hours"]
    S1 --> Alert["Alert on-call + optional\nauto-throttle or auto-revoke"]
    S2 --> Alert
    S3 --> Alert
    S4 --> Alert

    style Logs fill:#EEF0F7,stroke:#6366F1,color:#0F172A
    style Detect fill:#fff7ed,stroke:#f59e0b,color:#0F172A
    style Alert fill:#fef2f2,stroke:#dc2626,color:#0F172A

async def check_scope_probing(caller_id: str, window_minutes: int = 5) -> bool:
    """A burst of 403s from one caller in a short window often indicates
    an agent (or something manipulating it) trying tools outside its
    granted scope, worth surfacing even when every individual call was correctly denied."""
    recent_403s = await audit_store.count_outcomes(caller_id, outcome="authorization_denied", window_minutes=window_minutes)
    return recent_403s > 10

This closes Module 7. With prevention, containment, detection, and the audit log’s own hygiene in place, Module 8 moves to the client side: how Claude, other model providers, and open-weight models actually connect to and call the MCP servers built throughout this course.