MAF middleware — guardrails, logging, rate limits

Once you’ve built an agent that works, the next problems are the boring-but-critical ones: logging every call, blocking PII before it leaves your network, rate-limiting per user, killing runaway loops. MAF handles these with middleware — functions that wrap around agent invocations and tool calls, running before the agent receives input and after it produces output, without touching the agent’s own code. Think of middleware as a series of transparent sleeves stacked around the call: each sleeve can inspect, modify, or short- circuit the request as it passes through. If you’ve used ASP.NET, Express, or FastAPI middleware, the concept transfers directly.

Where middleware sits

There are two places middleware runs:

Both are functions (or classes) that receive a context object and a next callable. You inspect, optionally mutate, call next() to continue the chain, and inspect/mutate the result on the way back out.

A guardrail that scrubs API keys

The most common production middleware: scan inputs for secrets before they hit the model. If a user pastes an API key into a chat, you do not want it logged, embedded, or echoed back.

# requires: pip install agent-framework (for the commented agent wiring)
# The scrub() function below is plain Python — the middleware wiring is the
# real MAF shape.

import re

# A regex for common secret shapes. Real production uses something like
# Microsoft Presidio or a managed secret scanner — this is the toy version.
SECRET_PATTERNS = [
    re.compile(r"sk-[a-zA-Z0-9]{20,}"),                 # OpenAI-style
    re.compile(r"AKIA[0-9A-Z]{16}"),                    # AWS access key
    re.compile(r"ghp_[a-zA-Z0-9]{36}"),                 # GitHub PAT
    re.compile(r"xox[baprs]-[a-zA-Z0-9-]+"),            # Slack token
]

def scrub(text: str) -> tuple[str, list[str]]:
    """Replace secrets with [REDACTED]. Returns (clean_text, hits)."""
    hits = []
    for pat in SECRET_PATTERNS:
        for m in pat.finditer(text):
            hits.append(m.group()[:6] + "…")  # log first 6 chars for audit
        text = pat.sub("[REDACTED]", text)
    return text, hits

# The middleware itself. Signature: (context, next) -> None.
async def secret_scrubber_middleware(context, next):
    # 1. Inspect/mutate inputs BEFORE the agent runs.
    for msg in context.messages:
        if msg.role == "user" and isinstance(msg.content, str):
            clean, hits = scrub(msg.content)
            if hits:
                print(f"audit: scrubbed {len(hits)} secret(s): {hits}")
                msg.content = clean

    # 2. Call the next thing in the chain (other middleware, then the agent).
    await next()

    # 3. Optionally inspect the response on the way out.
    last = context.result.messages[-1]
    if isinstance(last.content, str):
        last.content, _ = scrub(last.content)  # belt + suspenders

# Wire it in:
#   from agent_framework import ChatAgent
#   agent = ChatAgent(
#       chat_client=chat_client,
#       instructions="Help the user.",
#       middleware=[secret_scrubber_middleware],
#   )

# Sanity check on the scrub function itself (this DOES run):
sample = "My key is sk-abc123def456ghi789jkl0 — help me debug."
clean, hits = scrub(sample)
print(clean)
print(hits)

My key is [REDACTED] — help me debug.
['sk-abc…']

The pasted key is replaced with [REDACTED] before the model (or any log) ever sees it, and the audit list keeps only the first six characters (sk-abc…) — enough to trace which key leaked without storing the secret itself. The structure is the same for every middleware: inspect → call next → inspect the result. Mutate the context to change inputs; mutate the result to change outputs.

Other middleware you’ll write

Logging / telemetry. Log every invocation with user_id, agent_name, input length, output length, latency, token cost. MAF emits OpenTelemetry spans automatically, but custom middleware is where you attach business-specific attributes (tenant, feature flag, A/B variant).

Rate limiting. Check Redis for count(user_id, minute) before calling next(). Raise or return a “rate limit exceeded” response if over the cap. This needs to run early in the chain so rate-limited requests don’t waste downstream work.

Termination / turn cap. For workflows or multi-turn agents, count turns in the context and short-circuit after N. Catches runaway loops that other middleware missed.

Cost guard. Track token spend per user per day. If a user crosses their budget, return a polite “quota exhausted” message instead of calling the model.

Function middleware for auditing tool calls. Every time the delete_user tool runs, write to an immutable audit log: (timestamp, user_id, tool_name, args, approved_by). This is the forensic record you’ll want if someone asks “why was this account deleted?” six months later.

Order matters

Middleware composes in the order you pass it — first one wraps the outside, last one wraps closest to the agent. Mental model: like nested function calls.

agent = ChatAgent(
    chat_client=chat_client,
    middleware=[
        auth_middleware,           # outermost — reject before anything else runs
        rate_limit_middleware,     # cheap, fast — fail-fast for limited users
        secret_scrubber_middleware,
        logging_middleware,        # logs the (already scrubbed) input
        # ... agent runs here ...
    ],
)

The rules of thumb:

1. Cheap, fail-fast middleware first. Auth and rate-limiting reject requests before you spend cycles on the rest.
2. Mutating middleware before logging. Otherwise your logs capture the pre-mutation state and your audit trail is wrong.
3. Cost/quota guards close to the agent. They need accurate estimates of what’s about to be called.

Function middleware — same idea, narrower scope

Function middleware wraps tool calls instead of the whole agent run. Useful when you want a guardrail that only applies to a subset of tools.

async def database_write_audit(context, next):
    """Audit every database mutation tool call."""
    if context.tool_name.startswith("db_write_"):
        audit_log.write(
            user=context.user_id,
            tool=context.tool_name,
            args=context.arguments,
            ts=datetime.utcnow(),
        )
    await next()

You attach function middleware similarly — typically via a function_middleware=[...] parameter on the agent or workflow.

In one breath

• Middleware wraps agent runs and tool calls — (context, next) functions that inspect/mutate input, call next(), then inspect/mutate the result, without touching agent code.
• Two scopes: agent middleware (per agent.run() — auth, rate limits, response logging) and function middleware (per tool call — tool guardrails, audit logs).
• Order matters — middleware nests like function calls: cheap fail-fast (auth, rate limit) first, mutating middleware before logging, cost/quota guards closest to the agent.
• The canonical guardrail is a secret scrubber on both input and output (defense in depth) — it must run before the logger so raw secrets never hit your logs.
• Test middleware like tools — an over-eager scrubber or rate limiter silently breaks legit requests.

Quick check

Next

That’s the MAF series. Next up: Google’s Agent Development Kit (ADK), the Gemini-native alternative — same problems, different ergonomics.