* Python: add ATR validation FunctionMiddleware sample (execution-boundary validation, #5366) Adds python/samples/02-agents/middleware/atr_validation_middleware.py: a FunctionMiddleware that validates tool arguments at the execution boundary and raises MiddlewareTermination before call_next() when they match an attack pattern, so the tool never runs. This is the deterministic, single-enforcement- point pattern named in #5366 and answers its open follow-up about a recommended validation-at-execution-boundary sample. The check is a small self-contained deny-list mirroring Agent Threat Rules (ATR) intent (prompt injection, exfiltration, credential access in tool args); a docstring notes how to swap in the full open ruleset via pyatr. No external dependency, so the sample stays import-clean. Updates the middleware README Files table. Signed-off-by: Adam Lin <adam@agentthreatrule.org> * Python: Samples: run the real ATR engine in atr_validation_middleware Address review on #6528: - Load and run the real ATR ruleset via pyatr (ATREngine + AgentEvent tool_call event) instead of re-implementing a regex deny-list; the built-in deny-list is now only a fallback when pyatr is not installed. - Add re.DOTALL (and a whole-text scan) to the fallback patterns so multiline injection payloads are not missed. - Move load_dotenv() into main() so importing the module has no side effects. - Route the middleware block/allow messages through a module logger instead of print(). - Include the matched ATR rule id in the log and in the MiddlewareTermination message for auditability. - Update the middleware README entry to match. * fix(samples): make ATR validation middleware pass ty/pyrefly typing CI Resolve the three type-checker errors flagged on the samples typing jobs (ty + pyrefly, reportMissingImports/reportAttributeAccessIssue via pyright): - pyatr is an optional, unstubbed runtime dependency that is not installed in the typing CI env; mark its imports with `# type: ignore` so the unresolved-import error is suppressed while keeping the graceful ImportError -> deny-list fallback intact. - Replace the function-attribute engine cache (`_detect_with_atr._engine`), which ty/pyrefly reject, with a clean `functools.lru_cache`-backed `_load_atr_engine()` loader. - Type the argument-scanning helpers to accept the real `FunctionInvocationContext.arguments` type (`BaseModel | Mapping[str, Any]`) and normalise a pydantic model via `model_dump()` before scanning, fixing the invalid-argument-type error. ty / pyrefly / pyright (samples config) / ruff check + format all clean on the file; runtime block/allow behaviour verified for both dict and BaseModel arguments. * Python: Samples: simplify ATR middleware to plain pyatr import Address review feedback (@eavanvalkenburg): now that the sample runs the real pyatr engine, drop the optional-import scaffolding. - Add a dependency header declaring pyatr (pip install pyatr). - Switch to a plain top-level `import pyatr` and remove the try/except ImportError fallback path. - Remove the regex deny-list (_FALLBACK_PATTERNS, _detect_with_fallback); keep 2-3 representative pattern shapes inline as a reference comment so readers still see the kind of rules ATR encodes. Detection is now a single straight-line engine call. - Keep the prior typing fixes: `# type: ignore` on the pyatr import (unstubbed, absent in the typing CI env), the functools.lru_cache engine loader, and the BaseModel | Mapping[str, Any] signatures. * fix: use PEP 723 inline script metadata for sample dependencies --------- Signed-off-by: Adam Lin <adam@agentthreatrule.org> Co-authored-by: eeee2345 <eeee2345@users.noreply.github.com> Co-authored-by: Eduard van Valkenburg <eavanvalkenburg@users.noreply.github.com>
Middleware samples
This folder contains focused middleware samples for Agent, chat clients, tools, sessions, and runtime context behavior.
Files
| File | Description |
|---|---|
agent_and_run_level_middleware.py |
Demonstrates combining agent-level and run-level middleware. |
agent_loop_middleware_refinement.py |
Demonstrates AgentLoopMiddleware with a should_continue predicate: a completion-marker refinement loop with feedback tracking and fresh_context. |
agent_loop_middleware_todos.py |
Demonstrates AgentLoopMiddleware with a should_continue predicate built from a TodoProvider via todos_remaining, so the agent keeps working while open todos remain. |
agent_loop_middleware_judge.py |
Demonstrates AgentLoopMiddleware.with_judge: a ChatClient judge re-runs the agent until it decides the original request was answered, with criteria shared between the agent and the judge. |
agent_loop_middleware_report.py |
Demonstrates composing two AgentLoopMiddleware on one agent: an inner todos_remaining loop that drafts a report todo-by-todo, wrapped by an outer report-style with_judge loop that re-runs it until an editor chat client judges the report publication-ready. |
atr_validation_middleware.py |
Demonstrates deterministic validation at the tool-execution boundary: a FunctionMiddleware that inspects the validated tool arguments and raises MiddlewareTermination before the tool runs when they match an attack rule. Loads the open, MIT-licensed Agent Threat Rules ruleset and runs the real engine locally (pip install pyatr), with a built-in deny-list fallback when it is not installed. |
chat_middleware.py |
Shows class-based and function-based chat middleware that can observe, modify, and override model calls. |
class_based_middleware.py |
Shows class-based agent and function middleware. |
decorator_middleware.py |
Demonstrates middleware registration with decorators. |
exception_handling_with_middleware.py |
Shows how middleware can handle failures and recover cleanly. |
function_based_middleware.py |
Shows function-based agent and function middleware. |
middleware_termination.py |
Demonstrates stopping a middleware pipeline early. |
override_result_with_middleware.py |
Shows how middleware can replace regular and streaming results, then post-process the final response. |
runtime_context_delegation.py |
Demonstrates delegating arguments with runtime context data. |
session_behavior_middleware.py |
Shows how middleware interacts with session-backed runs. |
shared_state_middleware.py |
Demonstrates sharing mutable state across middleware invocations. |
usage_tracking_middleware.py |
Demonstrates one chat middleware function that tracks per-call usage in non-streaming and streaming tool-loop runs. |
Running the usage tracking sample
The new usage tracking sample uses OpenAIChatClient, so set the usual OpenAI responses environment variables first:
export OPENAI_API_KEY="your-openai-api-key"
export OPENAI_CHAT_MODEL="gpt-4.1-mini"
Then run:
uv run samples/02-agents/middleware/usage_tracking_middleware.py
The sample forces a tool call so you can see middleware output for each inner model call in both non-streaming and streaming modes.
Security Considerations
AgentLoopMiddleware.with_judge (used by agent_loop_middleware_judge.py and
agent_loop_middleware_report.py) is an explicit opt-in to sending the original request and the
agent's latest response to a second, external judge chat client on every iteration. A compromised
or malicious judge endpoint could exfiltrate that data, or return a manipulated verdict/gap
analysis that gets fed back into the loop as feedback — a form of indirect prompt injection. Only
configure a judge client that points at a service you trust as much as the primary model.