Python: Packages: add agent-framework-atr (ATR validation middleware provider)

python/packages/atr/LICENSE

@@ -0,0 +1,21 @@
+    MIT License
+
+    Copyright (c) Microsoft Corporation.
+
+    Permission is hereby granted, free of charge, to any person obtaining a copy
+    of this software and associated documentation files (the "Software"), to deal
+    in the Software without restriction, including without limitation the rights
+    to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+    copies of the Software, and to permit persons to whom the Software is
+    furnished to do so, subject to the following conditions:
+
+    The above copyright notice and this permission notice shall be included in all
+    copies or substantial portions of the Software.
+
+    THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+    IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+    FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+    AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+    LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+    OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+    SOFTWARE

python/packages/atr/README.md

@@ -0,0 +1,94 @@
+## Microsoft Agent Framework – ATR Integration (Python)
+
+`agent-framework-atr` adds deterministic [Agent Threat Rules (ATR)](https://github.com/Agent-Threat-Rule/agent-threat-rules) validation to the Microsoft Agent Framework. It lets you block malicious tool calls and user inputs using an open, MIT-licensed detection ruleset for AI-agent threats — prompt injection, tool-argument tampering, credential exfiltration, and more.
+
+> Status: **Preview**
+
+ATR is an independent, open standard. This package is a thin integration that runs the upstream [`pyatr`](https://pypi.org/project/pyatr/) engine inside the Agent Framework middleware pipeline. Detection is fully local and deterministic — there is no model call in the enforcement path, so block/allow decisions are reproducible and auditable.
+
+### Key Features
+
+- Deterministic enforcement at the tool-execution boundary (`ATRFunctionMiddleware`) — the pattern recommended in issue #5366: the check runs before the tool executes, so a matched call never fires.
+- Inbound input scanning at the agent boundary (`ATRAgentMiddleware`) — blocks a run when a user message matches a rule.
+- Works with any `Agent` using the standard Agent Framework middleware pipeline.
+- No external service, credentials, or network calls — the ruleset ships with `pyatr` and runs in-process.
+- `audit_only` (shadow) mode: record and log matches without blocking.
+
+### When to Use
+
+Add ATR when you want a deterministic, auditable guard that:
+
+- Blocks prompt-injection or exfiltration payloads that land in tool arguments before the tool runs.
+- Rejects malicious user input before it reaches the model.
+- Applies a maintained, community-driven ruleset without hand-rolling deny-lists.
+
+---
+
+## Quick Start
+
+```python
+import asyncio
+
+from agent_framework import Agent
+from agent_framework.openai import OpenAIChatCompletionClient
+from agent_framework_atr import ATRFunctionMiddleware
+
+
+async def main() -> None:
+    client = OpenAIChatCompletionClient()
+
+    agent = Agent(
+        client=client,
+        instructions="You are a helpful assistant.",
+        tools=[...],
+        middleware=[ATRFunctionMiddleware()],
+    )
+
+    # A tool call whose arguments match an ATR rule is blocked before the tool runs;
+    # the middleware raises MiddlewareTermination with the matched rule id.
+    result = await agent.run("What's the weather in Tokyo?")
+    print(result)
+
+
+asyncio.run(main())
+```
+
+To guard the inbound user message instead of (or in addition to) tool arguments:
+
+```python
+from agent_framework_atr import ATRAgentMiddleware
+
+agent = Agent(client=client, instructions="...", middleware=[ATRAgentMiddleware()])
+```
+
+---
+
+## Configuration
+
+Both middleware share the same options:
+
+```python
+from agent_framework_atr import ATRDetector, ATRFunctionMiddleware
+
+# Share a single detector so the ruleset is loaded once.
+detector = ATRDetector(
+    rules_dir=None,              # None: use the ruleset bundled with pyatr; or a path to your own ATR rules
+    min_severity="informational",  # only act on matches at or above this severity
+)
+
+middleware = ATRFunctionMiddleware(
+    detector=detector,
+    audit_only=False,            # True: record + log matches without blocking (shadow mode)
+)
+```
+
+On a block, the matched detection is recorded on `context.metadata["atr_detection"]` (an `ATRDetection` with `rule_id`, `severity`, `confidence`, and `title`) for later inspection by downstream middleware or logging.
+
+---
+
+## Notes
+
+- **Deterministic**: detection runs the ATR engine locally over the text; no model is called to make the block/allow decision.
+- **Severity threshold**: `min_severity` filters weaker matches; the highest-severity match is the one reported.
+- **Audit mode**: use `audit_only=True` to measure what would be blocked before enforcing.
+- **Ruleset**: `pyatr` ships the published ATR ruleset; point `rules_dir` at your own directory to run a custom or pinned set.

python/packages/atr/agent_framework_atr/__init__.py

@@ -0,0 +1,11 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from ._engine import ATRDetection, ATRDetector
+from ._middleware import ATRAgentMiddleware, ATRFunctionMiddleware
+
+__all__ = [
+    "ATRAgentMiddleware",
+    "ATRDetection",
+    "ATRDetector",
+    "ATRFunctionMiddleware",
+]

python/packages/atr/agent_framework_atr/_engine.py

@@ -0,0 +1,95 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+"""Local, deterministic ATR detection backed by the ``pyatr`` engine.
+
+The engine is loaded once and reused. Detection runs entirely in-process with
+no model call, so block/allow decisions are reproducible and auditable.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Any
+
+import pyatr
+
+# Severity ranks used to filter matches against ``min_severity`` and to keep the
+# highest-severity hit. Mirrors the ATR schema severity enum.
+_SEVERITY_ORDER: dict[str, int] = {
+    "informational": 0,
+    "low": 1,
+    "medium": 2,
+    "high": 3,
+    "critical": 4,
+}
+
+
+@dataclass(frozen=True)
+class ATRDetection:
+    """A single ATR rule match.
+
+    Attributes:
+        rule_id: The matched rule identifier, e.g. ``ATR-2026-00001``.
+        severity: The rule severity (critical, high, medium, low, informational).
+        confidence: The rule confidence label reported by the engine.
+        title: The human-readable rule title.
+    """
+
+    rule_id: str
+    severity: str
+    confidence: str
+    title: str
+
+
+class ATRDetector:
+    """Loads an ATR ruleset once and evaluates agent text against it.
+
+    Detection is delegated to the upstream ``pyatr`` engine and is fully local
+    and deterministic. Construct one detector and share it across middleware to
+    avoid reloading rules on every call.
+
+    Args:
+        rules_dir: Directory of ATR rule YAML files to load. When ``None`` the
+            ruleset bundled with ``pyatr`` is used.
+        min_severity: Minimum severity a match must have to be reported. One of
+            critical, high, medium, low, informational. Defaults to
+            ``informational`` (report every match).
+    """
+
+    def __init__(self, *, rules_dir: str | None = None, min_severity: str = "informational") -> None:
+        engine: Any = pyatr.ATREngine()
+        if rules_dir is None:
+            engine.load_default_rules()
+        else:
+            engine.load_rules_from_directory(rules_dir)
+        self._engine: Any = engine
+        self._min_rank: int = _SEVERITY_ORDER.get(min_severity.lower(), 0)
+
+    def detect(self, text: str, *, event_type: str = "llm_input", field: str = "user_input") -> ATRDetection | None:
+        """Evaluate ``text`` and return the highest-severity match, or ``None``.
+
+        Args:
+            text: The agent text to scan (user input, tool arguments, ...).
+            event_type: The ATR event type to evaluate the text as, e.g.
+                ``llm_input`` or ``tool_call``.
+            field: The ATR field name the text is exposed under so field-scoped
+                rule conditions (e.g. ``tool_args``) can match.
+
+        Returns:
+            The highest-severity :class:`ATRDetection` at or above
+            ``min_severity``, or ``None`` when nothing matches.
+        """
+        if not text:
+            return None
+        event: Any = pyatr.AgentEvent(content=text, event_type=event_type, fields={field: text})
+        matches: Any = self._engine.evaluate(event)
+        # ``evaluate`` returns matches sorted critical-first.
+        for match in matches:
+            if _SEVERITY_ORDER.get(str(match.severity).lower(), 0) >= self._min_rank:
+                return ATRDetection(
+                    rule_id=str(match.rule_id),
+                    severity=str(match.severity),
+                    confidence=str(match.confidence),
+                    title=str(match.title),
+                )
+        return None