Build a Runtime Agent-Security Monitoring Workflow with Microsoft Defender for AI Agents

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

The security industry spent 2025 protecting models. The market is now converging on the harder problem: protecting agents at runtime. Microsoft's real-time agent protection for Defender for Cloud Apps, in preview in 2026, is the clearest signal yet that the agentic loop itself — the sequence of tool invocations, data reads, and actions an agent performs after the model answers — has become the security boundary. Prompt filters and model guardrails cannot help here, because by the time the agent is calling a tool, the model has already answered; the only place left to enforce policy is in the execution layer. This dispatch builds a LangGraph monitoring workflow that sits on that boundary: it inspects agent tool calls at runtime, enforces allowlists, detects anomalous actions, blocks what it must, auto-remediates what it can, and hands everything over to incident response with the evidence intact.

What Real-Time Agent Protection Actually Evaluates

Defender's real-time protection inspects AI agent activity throughout the agentic loop and blocks risky actions before they execute. It evaluates the parts of the loop that old tooling never saw: the user request, the agent's response, the tool invocation, and the tool response. For Agent 365-managed agents it integrates with Work IQ MCP to evaluate customer MCP tool invocations before they run; Copilot Studio and Foundry agents get the same evaluation in preview; local agents are covered by AI agent runtime protection in Defender for Endpoint, which hooks Claude Code, Codex CLI, and GitHub Copilot CLI at their native event points.

Two rule types do the enforcement. A built-in default rule audits all agents, recording matching activity as a behavior without stopping the action — this is how you get visibility before you trust blocking. Custom rules block matching actions before they execute and are scoped to specific agents, with detection types such as secret exfiltration, malicious content propagation, evasion techniques, unsafe email domains, jailbreak attempts, and indirect prompt injection (XPIA). Every audit or block event lands in the BehaviorInfo table with what happened, why it was considered risky, and which agent, user, and tool were involved. You can query that table in Advanced Hunting with KQL and wire it into downstream automation — exactly what our workflow does. For the surrounding automation patterns, the Daily AI World workflows library is tracking runtime-guard rails like this as they go from preview to GA.

Architecture: The Guarded Agent Loop

The workflow wraps your agent's tool execution in a pre-tool gate and a post-tool verification node. LangGraph provides the state and the conditional routing; Defender provides the runtime decisions; your allowlist, egress, and credential-scope policies provide the deterministic local checks that catch what preview telemetry has not yet learned.

   ┌───────────────────────── AGENTIC LOOP ─────────────────────────┐
   │                                                                │
   │  ┌──────────┐    ┌──────────────┐    ┌──────────────────────┐  │
   │  │ User     │───►│ Model / Agent │───►│  Tool Invocation     │  │
   │  │ request  │    │ decision      │    │  (pre-tool gate)     │  │
   │  └──────────┘    └──────────────┘    └──────────┬───────────┘  │
   │            ▲                                    │              │
   │            │                                    ▼              │
   │            │                       ┌──────────────────────┐   │
   │            │                       │ Defender Real-Time    │   │
   │            │                       │ Protection (audit/    │   │
   │            │                       │ block) · allowlist ·  │   │
   │            │                       │ egress · credential   │   │
   │            │                       └──────────┬───────────┘   │
   │            │                                  │              │
   │            │                ┌─────────────────┴──────────┐   │
   │            │                │ ALLOW                      │   │
   │            │                ▼                            │   │
   │            │          ┌──────────┐        ┌─────────────────┐│
   │            │          │ Execute  │        │  Anomaly +      ││
   │            │          │ tool     │        │  egress check   ││
   │            │          └──────────┘        └────────┬────────┘│
   │            └────────── response ▲                  │         │
   │                              │  ┌───────────────────┘         │
   │                              │  │  BLOCK / FLAG               │
   │                              │  ▼                            │
   └──────────────────────────────┼────────────────────────────────┘
                                   │
                    ┌──────────────┴──────────────┐
                    ▼                             ▼
        ┌─────────────────────┐       ┌─────────────────────────┐
        │  Auto-Remediation    │       │  Incident Response      │
        │  revoke scope ·      │       │  BehaviorInfo → SOC     │
        │  egress deny ·       │       │  KQL hunting · alert    │
        │  session terminate   │       └─────────────────────────┘
        └─────────────────────┘

The gate never trusts the model's own claim about what a tool does. Every call is evaluated by the policy stack before execution, and the response is verified again after execution for exfiltration-shaped anomalies.

Prerequisites

Enable security for AI agents in Defender, connect the platform that runs your agents (Agent 365, Copilot Studio, or Foundry), and confirm real-time protection is on. In the Defender portal: Settings — Security for AI — Policies & rules — Real-time protection. Start with the built-in Default (audit) rule so behaviors accumulate in BehaviorInfo before you write blocking rules. For local agents, enable AI agent runtime protection in Defender for Endpoint.

Environment Configuration

.env holds Defender credentials and the policy knobs:

# .env
DEFENDER_TENANT_ID=00000000-0000-0000-0000-000000000000
DEFENDER_CLIENT_ID=11111111-1111-1111-1111-111111111111
DEFENDER_CLIENT_SECRET=...
AGENT_ID=agents/my-agent              # Entra agent ID scope
POLICY_MODE=audit                     # audit first, then block
ALLOWLIST_TOOLS=read_repo,search_web,create_ticket
EGRESS_ALLOWED=api.github.com,api.atlassian.com
CREDENTIAL_SCOPES=repo:read,issues:write
BLOCK_WEBHOOK_URL=https://hooks.slack.com/.../security
RETRY_MAX_ATTEMPTS=3

Never commit this file; the client secret is a live credential.

Schema Definitions

schemas.py models tool calls, decisions, and the behavior record that mirrors BehaviorInfo:

# schemas.py
from __future__ import annotations

from enum import Enum
from typing import Optional

from langgraph.graph import MessagesState
from pydantic import BaseModel, Field


class Decision(str, Enum):
    ALLOW = "allow"
    BLOCK = "block"
    FLAG = "flag"


class ToolCall(BaseModel):
    tool: str
    arguments: dict = Field(default_factory=dict)
    egress_host: Optional[str] = None
    credential_scope: Optional[str] = None
    session_id: str


class ToolCallDecision(BaseModel):
    call: ToolCall
    decision: Decision
    rule: str
    reason: str


class SecurityState(MessagesState):
    agent_id: str
    session_id: str
    policy_mode: str = "audit"
    pre_decision: Optional[ToolCallDecision] = None
    post_decision: Optional[ToolCallDecision] = None
    tool_result: Optional[str] = None
    behaviors: list[dict] = Field(default_factory=list)
    incident_id: Optional[str] = None

Guard Tools

tools.py provides three LangChain tools: the Defender evaluation wrapper, the local allowlist/egress/credential check, and the remediation handoff:

# tools.py
from __future__ import annotations

import os

import httpx
from langchain_core.tools import tool

TENANT = os.getenv("DEFENDER_TENANT_ID")
CLIENT_ID = os.getenv("DEFENDER_CLIENT_ID")
SECRET = os.getenv("DEFENDER_CLIENT_SECRET")


def _bearer() -> str:
    # Client-credentials token for the Defender API.
    resp = httpx.post(
        f"https://login.microsoftonline.com/{TENANT}/oauth2/v2.0/token",
        data={
            "grant_type": "client_credentials",
            "client_id": CLIENT_ID,
            "client_secret": SECRET,
            "scope": "https://api.security.microsoft.com/.default",
        },
    )
    resp.raise_for_status()
    return resp.json()["access_token"]


@tool
def evaluate_tool_call(tool: str, arguments: dict) -> dict:
    """Evaluate a tool invocation through Defender real-time protection."""
    resp = httpx.post(
        "https://api.security.microsoft.com/api/agents/real-time-protection/evaluate",
        json={"agent_id": os.getenv("AGENT_ID"), "tool": tool, "arguments": arguments},
        headers={"Authorization": f"Bearer {_bearer()}"},
        timeout=30,
    )
    resp.raise_for_status()
    return resp.json()


@tool
def enforce_policy(call: dict) -> dict:
    """Deterministic local checks: allowlist, egress, credential scope."""
    violations = []
    if call["tool"] not in os.getenv("ALLOWLIST_TOOLS", "").split(","):
        violations.append(f"tool {call['tool']} not allowlisted")
    host = (call.get("egress_host") or "").lower()
    if host and host not in os.getenv("EGRESS_ALLOWED", "").split(","):
        violations.append(f"egress to {host} denied")
    scope = call.get("credential_scope") or ""
    allowed = os.getenv("CREDENTIAL_SCOPES", "").split(",")
    if scope and not set(scope.split()).issubset(allowed):
        violations.append(f"credential scope {scope} exceeds granted scopes")
    return {"violations": violations}


@tool
def auto_remediate(agent_id: str, session_id: str, reason: str) -> dict:
    """Revoke credential scope, terminate the session, and notify SOC."""
    revoke = httpx.post(
        "https://api.security.microsoft.com/api/agents/remediate",
        json={"agent_id": agent_id, "session_id": session_id, "reason": reason},
        headers={"Authorization": f"Bearer {_bearer()}"},
        timeout=30,
    )
    revoke.raise_for_status()
    httpx.post(os.getenv("BLOCK_WEBHOOK_URL"), json={"event": "agent_blocked", "reason": reason})
    return {"revoked": True, "session": session_id}

The LangGraph Guard Graph

graph.py builds the guarded loop. The pre-tool gate returns an allow/block/flag decision and routes accordingly:

# graph.py
from __future__ import annotations

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, StateGraph

from schemas import Decision, SecurityState, ToolCall, ToolCallDecision
from tools import auto_remediate, enforce_policy, evaluate_tool_call


def _pre_tool_gate(state: SecurityState) -> dict:
    call = state.messages[-1]["tool_call"]
    local = enforce_policy.invoke({"call": call})
    remote = evaluate_tool_call.invoke(tool=call["tool"], arguments=call["arguments"])

    if local["violations"] or remote.get("action") == "block":
        decision = Decision.BLOCK
        rule = "allowlist" if local["violations"] else "defender_block"
        reason = "; ".join(local["violations"] or [remote.get("reason", "")])
    elif remote.get("action") == "flag":
        decision = Decision.FLAG
        rule = "anomaly"
        reason = remote.get("reason", "flagged by Defender")
    else:
        decision = Decision.ALLOW
        rule = "pass"
        reason = "policy satisfied"

    if state.policy_mode == "audit":
        decision = Decision.ALLOW  # observe before you enforce

    return {"pre_decision": ToolCallDecision(call=ToolCall(**call), decision=decision, rule=rule, reason=reason)}


def _route_gate(state: SecurityState) -> str:
    if state.pre_decision.decision == Decision.ALLOW:
        return "execute"
    return "remediate"


def _execute(state: SecurityState) -> dict:
    # The real tool runs here; result is kept for post-tool verification.
    result = run_agent_tool(state.pre_decision.call.tool, state.pre_decision.call.arguments)
    return {"tool_result": result}


def _post_tool_check(state: SecurityState) -> dict:
    # Detect exfiltration-shaped anomalies in the tool response.
    anomalies = detect_anomalies(state.pre_decision.call, state.tool_result)
    if anomalies and state.policy_mode == "block":
        return {"post_decision": ToolCallDecision(call=state.pre_decision.call, decision=Decision.BLOCK, rule="exfil_scan", reason=anomalies)}
    return {"post_decision": ToolCallDecision(call=state.pre_decision.call, decision=Decision.ALLOW, rule="pass", reason="clean")}


def _remediate(state: SecurityState) -> dict:
    result = auto_remediate.invoke(
        agent_id=state.agent_id,
        session_id=state.session_id,
        reason=state.pre_decision.reason,
    )
    return {"incident_id": result.get("incident_id"), "behaviors": [{"event": "block", "rule": state.pre_decision.rule, "reason": state.pre_decision.reason}]}


def build_guard() -> StateGraph:
    g = StateGraph(SecurityState)
    g.add_node("pre_gate", _pre_tool_gate)
    g.add_node("execute", _execute)
    g.add_node("post_check", _post_tool_check)
    g.add_node("remediate", _remediate)
    g.set_entry_point("pre_gate")
    g.add_conditional_edges("pre_gate", _route_gate, {"execute": "execute", "remediate": "remediate"})
    g.add_edge("execute", "post_check")
    g.add_edge("post_check", END)
    g.add_edge("remediate", END)
    return g

Entry Point

main.py runs the guarded loop against an agent session:

# main.py
import asyncio

from dotenv import load_dotenv

from graph import build_guard
from schemas import SecurityState

load_dotenv()


async def main() -> None:
    graph = build_guard().compile(checkpointer=InMemorySaver())
    state = SecurityState(
        agent_id="agents/my-agent",
        session_id="sess-77",
        policy_mode="block",
        messages=[{"role": "user", "content": "send the latest sales report", "tool_call": {
            "tool": "email_send", "arguments": {"to": "buyer@example.com"},
            "egress_host": "smtp.example.com", "credential_scope": "mail:send",
        }}],
    )
    result = await graph.ainvoke(state, config={"configurable": {"thread_id": "sess-77"}})
    print(result.get("pre_decision"))
    print(result.get("incident_id"))


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

Retry Rules

The important design rule: the pre-tool gate fails closed. In a security graph, a timeout is a block, not a pass — that asymmetry is what separates runtime protection from a speed bump.

Audit vs Block: When to Flip the Switch

Run audit for at least a week, measure false-positive rate against BehaviorInfo, and only then promote high-confidence detection types to block mode.

Operating Notes

Three realities govern this workflow. First, coverage is platform-dependent: tool-invocation evaluation is strongest for Agent 365 agents on Work IQ MCP, in preview for Copilot Studio and Foundry, and a different (endpoint) mechanism for local CLI agents — design your allowlist checks so they work even where Defender telemetry is still maturing. Second, session context is the key: scope every decision to session_id so remediation can revoke a credential scope and terminate a session without killing the whole fleet. Third, prompt evidence is sensitive: enable prompt evidence collection with redaction on, and treat the snippets as PII-adjacent in your audit pipeline. Query BehaviorInfo, CloudAppEvents, and AlertInfo with KQL for hunting; the schema is stable enough to build detections on today.

Extend the handoff: point BLOCK_WEBHOOK_URL at your SIEM or SOAR so every block event opens a ticket with the behavior record attached. The same pattern wraps other MCP-connected tooling, and Microsoft continues to expand supported platforms — watch the latest AI news page as previews convert to GA.

FAQ

What does Microsoft Defender's real-time agent protection cover?

It evaluates AI agent activity throughout the agentic loop — user requests, agent responses, tool invocations, and tool responses — and blocks risky actions before they execute. Coverage spans Agent 365 tool invocations via Work IQ MCP, Copilot Studio and Foundry agents in preview, and local agents through Defender for Endpoint runtime protection.

What is the difference between the default rule and custom rules?

The default rule audits all agents, recording matching activity as a behavior without stopping it. Custom rules block matching actions before they execute and are scoped to specific agents with exclusions, letting you enforce only high-confidence detection types.

Where do block and audit events go?

They are recorded as behaviors in the BehaviorInfo table, including what happened, why it was considered risky, and which agent, user, and tool were involved. You query them with KQL in Advanced Hunting alongside AlertInfo and CloudAppEvents.

How does the LangGraph gate fail closed?

The pre-tool gate treats a Defender API timeout or evaluation failure as a BLOCK decision and routes to remediation, never silently passing the tool call. The POLICY_MODE flag is what relaxes it to audit while you tune rules.

What should I remediate automatically?

Revoke the credential scope that the anomalous call used, terminate the offending session, and notify SOC with the full behavior record. Do not auto-remediate across sessions or agents — that is the incident response handoff's job, not a graph node's.