Your Agent Runs Whatever the Model Decides. Middleware Is How You Watch Every Step.

Three interception layers, one call-chain contract, and the concrete patterns for blocking, overriding, and logging every run, tool call, and model call an agent makes.

You cannot fix, log, or block what you cannot see inside an agent's run. Middleware is the one mechanism that gives you that visibility without rewriting the agent.

In this article: You will learn the three layers Microsoft Agent Framework gives you for intercepting agent behavior (agent, function, and chat middleware), the call-chain pattern every middleware function follows, and the specific moves you can make from inside one: terminating a run early, overriding its result, sharing state across instances, and routing a runtime value to the surface that actually needs it. The article closes by wiring a logging-and-validation guard onto a market-research agent that blocks a bad request before it ever reaches the model.

An agent that calls a model, gets back a tool call, executes it, and calls the model again, sometimes several times before it produces an answer, does not leave much of a trail by default. Ask it why it did what it did, and the honest answer is that nobody logged it. Ask it to refuse a certain class of request before spending a model call on it, and the honest answer is that nothing in the loop was built to check first. That gap, between what an agent does and what you can observe or control about it, is exactly where middleware lives.

Middleware is a familiar idea borrowed from web frameworks and applied to agent behavior: a piece of code that wraps something else without living inside it. Logging, validation, rate limiting, retry logic, and exception handling all share the same shape. Something needs to happen around a piece of behavior, not embedded in it, so that the underlying logic stays untouched while the behavior around it changes.

Microsoft Agent Framework gives you three distinct places to hook that logic in: around a whole run, around a single tool call, and around a single call to the model itself. This article walks all three, the call-chain pattern every middleware function follows, the specific things you can do from inside one, and a concrete example: a logging-and-validation pair that blocks a request before it ever reaches the model.

Three layers, one job

Agent Framework gives you three distinct places to hook in, and each one intercepts a different piece of the agent's work.

Agent middleware wraps a whole run: everything from the messages coming in to the response going out, tool calls and all. Function middleware wraps a single tool invocation: one call to one function, with access to its arguments and its return value. Chat middleware wraps a single call to the model itself, the raw messages and options going to the inference service and the response coming back.

That last one matters more than it sounds like it should. A tool-calling run rarely makes exactly one model call. It calls the model, gets back a tool call, executes the tool, and calls the model again with the tool result, sometimes several times before it produces a final answer. Chat middleware runs on every one of those model calls, not once per agent run, which makes it the right layer for anything that needs to see what is actually being sent to the model turn by turn rather than what the agent returns at the end.

All three layers share the same contract: when you register more than one middleware of the same type, they form a chain, and each middleware is expected to call the next one to keep the chain moving. In Python that is an awaited call_next() callback with no arguments; the shared context object carries the state, and middleware mutates it directly instead of passing anything back through the callback. Here is the smallest possible version, logging middleware that wraps the whole run:

from agent_framework import AgentContext
from collections.abc import Awaitable, Callable

async def logging_agent_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    print("[Agent] Starting execution")
    await call_next()
    print("[Agent] Execution completed")

Everything before await call_next() runs before the agent executes; everything after runs once it is done. That is the whole pattern, whether you are logging, validating, or timing something. C# uses a next Func instead of a bare call_next(), and it hands you the inner agent directly rather than a shared context object:

async Task<AgentResponse> CustomAgentRunMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
    Console.WriteLine(response.Messages.Count);
    return response;
}

Same shape, different vocabulary: instead of mutating a context object and calling call_next(), you call innerAgent.RunAsync yourself and return whatever it gives back, or something else entirely if you are overriding the result.

Gotcha: C# function calling middleware only works with an AIAgent that uses FunctionInvokingChatClient under the hood, which ChatClientAgent does. If you have built a custom AIAgent implementation that does not route through that chat client, function calling middleware never fires.

Agent-level versus run-level: who gets touched, and when

Middleware can be registered two ways. Agent-level middleware is attached when you build the agent and applies to every single run for that agent's lifetime. Run-level middleware is passed to one specific run() call and only applies there.

The construction call below shows both scopes in one example: two agent-level middleware registered at build time, plus a run-level middleware passed only to the second call.

async with Agent(
    client=FoundryChatClient(credential=credential),
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=get_weather,
    middleware=[  # ①
        SecurityAgentMiddleware(),   # applies to every run
        TimingFunctionMiddleware(),  # applies to every run
    ],
) as agent:
    result1 = await agent.run("What's the weather in Seattle?")  # ②
    result2 = await agent.run(  # ③
        "What's the weather in Portland?",
        middleware=[logging_chat_middleware],  # this run only  ④
    )

① Two agent-level middleware are registered on the Agent constructor, so they wrap every run made through this agent for its whole lifetime. ② result1 only sees the two agent-level middleware; nothing here overrides or adds to that scope. ③ The second call reuses the same agent, so it still gets the agent-level pair wrapping it too. ④ Passing middleware= directly to run() scopes logging_chat_middleware to this call alone, layered inside the agent-level pair.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/01-agent-run-level-middleware.py shows the runnable form.

result1 only sees the two agent-level middleware. result2 sees those same two, plus logging_chat_middleware for that call alone. When both scopes are registered, agent-level middleware always wraps run-level middleware, which wraps the agent itself. For agent middleware [A1, A2] and run middleware [R1, R2], the order is A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1. Function and chat middleware follow the same wrapping principle when a tool call or model call happens inside that run.

C# gets the same two scopes through the same builder pattern, just applied at different points: once at construction time for agent-level, or inline right before a specific RunAsync call for run-level.

// Agent-level: registered once, applies to every run  ①
var agentWithMiddleware = baseAgent  // ②
    .AsBuilder()
        .Use(runFunc: SecurityMiddleware, runStreamingFunc: SecurityStreamingMiddleware)
    .Build();

// Run-level: built inline, applies only to this call  ③
Console.WriteLine(await baseAgent  // ④
    .AsBuilder()
        .Use(runFunc: DebugMiddleware, runStreamingFunc: DebugStreamingMiddleware)
    .Build()
    .RunAsync("What's the weather in Tokyo?"));

① Marks the construction path: middleware registered here applies to every future run of agentWithMiddleware. ② The wrapped agent is assigned to a variable, so other code can reuse this agent-level middleware chain. ③ Marks the run-level path: middleware built here applies only to the one RunAsync call that follows. ④ The wrapper is built, used, and discarded inline; nothing here is kept for reuse.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/02-agent-run-level-middleware.cs shows the runnable form.

The second block builds a fresh, lightweight wrapper around baseAgent and calls RunAsync on it directly, without ever assigning it to a variable other agents reuse. That is the run-level pattern in C#: build, use once, discard.

Reach for agent-level middleware when the behavior should apply unconditionally, security checks and logging are the common case. Reach for run-level when the behavior only makes sense for one call, a debug flag a caller sets on a single request, or a cache that only one code path wants.

Defining middleware: function-based, class-based, or decorated

Python supports three ways to write middleware, and they are interchangeable. Pick whichever fits the complexity of what you are building. A plain async function works for stateless logic:

async def inject_tool_runtime_defaults(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    context.function_invocation_kwargs.setdefault("tenant", "contoso")
    await call_next()

A class inheriting from AgentMiddleware, FunctionMiddleware, or ChatMiddleware works when the middleware needs to carry its own state, an accumulator, a cache, or a configuration object passed to __init__:

from agent_framework import AgentMiddleware, AgentContext

class LoggingAgentMiddleware(AgentMiddleware):
    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await call_next()
        print("[Agent Class] Execution completed")

The process method has the exact same signature as the function-based version: context and call_next in, None out. And if you are not using type annotations, or you want the framework to know a function's middleware type without inferring it, a decorator makes the intent explicit:

from agent_framework import agent_middleware

@agent_middleware
async def simple_agent_middleware(context, call_next):
    print("Before agent execution")
    await call_next()
    print("After agent execution")

C# only has the function-based shape, but it uses the same next-callback contract shown above, registered with .AsBuilder().Use(...). Go's public-preview SDK covers this territory too, but with one interface instead of three, agent.Middleware, chained in the order you register them on agent.Config.Middlewares. That single interface wraps the whole custom layer, context providers included, rather than giving you separate agent, function, and chat entry points the way C# and Python do.

Terminating a run before it starts

The most common thing middleware does beyond logging is decide a request should not proceed at all. In Python, you signal that by setting context.result to whatever you want returned, then raising MiddlewareTermination. Here is a pre-termination guard checking the last user message before the agent ever sees it:

from agent_framework import AgentContext, AgentResponse, Message, MiddlewareTermination

async def blocking_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    last_message = context.messages[-1] if context.messages else None  # ①
    if last_message and last_message.text and "blocked" in last_message.text.lower():  # ②
        context.result = AgentResponse(  # ③
            messages=[Message(role="assistant", contents=["This request was blocked by middleware."])]
        )
        raise MiddlewareTermination(result=context.result)  # ④

    await call_next()  # ⑤

① Reads the last message in the conversation, guarding against an empty list. ② The condition checks that message's text for the blocked keyword before doing anything else. ③ When it matches, the middleware builds the exact response it wants the caller to see. ④ Raising MiddlewareTermination stops the rest of the chain immediately instead of letting the agent run. ⑤ When the check does not match, the middleware falls through to call_next() and the run proceeds normally.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/03-blocking-middleware.py shows the runnable form.

Raising MiddlewareTermination stops the rest of the chain immediately and skips the normal execution path. You will also see a simpler variant in the wild that just returns without calling call_next() and without raising anything, which has the same practical effect of short-circuiting the chain, but only MiddlewareTermination is the documented, explicit signal for "stop here on purpose." This pattern works identically for agent, function, and chat middleware; only what context.result needs to hold changes.

C# does not have an exception-based termination signal. Agent run middleware terminates a run simply by not calling innerAgent.RunAsync and returning a different AgentResponse instead, the same guard clause pattern you would use anywhere else in C#. Function calling middleware has its own dedicated switch: set FunctionInvocationContext.Terminate = true to stop the function-calling loop from sending results back to the model.

Gotcha: terminating the C# function-calling loop mid-flight can leave chat history inconsistent, a function call message with no matching result. The docs call this out explicitly: that history may become unusable for further runs on the same session. If you need to stop a tool call from completing, prefer blocking it at the agent or function middleware layer before it starts, rather than terminating the loop after the model has already committed to calling it.

Overriding a result

Termination throws the response away entirely. Overriding lets the original logic run and then reshapes what comes back, useful for appending a disclaimer, redacting a field, or wrapping the response in a consistent envelope.

async def weather_override_middleware(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]]
) -> None:
    await call_next()  # ①

    if context.result is not None:  # ②
        if context.stream:  # ③
            async def override_stream():
                yield AgentResponseUpdate(contents=[Content.from_text(text="Weather override applied.")])
            context.result = override_stream()  # ④
        else:
            context.result = AgentResponse(  # ⑤
                messages=[Message(role="assistant", contents=["Weather override applied."])]
            )

① The original agent logic runs to completion first; there is nothing to override until call_next() returns. ② The override only applies when a result actually exists, so an already-terminated run is left alone. ③ context.stream tells the middleware which shape context.result has right now. ④ For a streaming run, the replacement must also be an async generator, so the middleware defines one and assigns it here. ⑤ For a non-streaming run, the replacement is a plain AgentResponse built the normal way.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/04-weather-override-middleware.py shows the runnable form.

context.stream is the flag that tells you which shape context.result actually has. For a non-streaming run, it is a complete AgentResponse. For a streaming run, it is an async generator yielding AgentResponseUpdate chunks, and replacing it means providing your own generator with the same shape. Check context.stream before you touch context.result, or your override middleware works in the demo and breaks the first time a caller streams.

C# does not have a separate streaming result type to branch on here; the return type of your middleware function already matches whichever RunAsync overload it is wired to. Building a new AgentResponse with modified messages and returning it is the whole pattern:

async Task<AgentResponse> ResultOverrideMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    var response = await innerAgent.RunAsync(messages, session, options, cancellationToken);  // ①
    var modified = response.Messages.Select(msg =>  // ②
        msg.Role == ChatRole.Assistant && msg.Text is not null  // ③
            ? new ChatMessage(ChatRole.Assistant, msg.Text + "\n\n_Disclaimer: AI-generated._")  // ④
            : msg).ToList();
    return new AgentResponse(modified);  // ⑤
}

① The original agent runs first, exactly as it would without this middleware. ② Building the replacement message list starts from the response's own messages. ③ Only assistant messages that actually carry text are candidates for the rewrite. ④ Matching messages get the disclaimer appended to their existing text. ⑤ The new AgentResponse built from the modified list is what the caller actually receives.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/05-result-override-middleware.cs shows the runnable form.

Every guard or rewrite middleware ends up choosing between three outcomes: stop the chain outright, let it run and reshape the result, or let it run untouched. The state diagram below lines up those three paths against the moment each one is decided.

Sharing state across middleware instances

Sometimes two middleware need to talk to each other: one counts something the other reports. In Python, the cleanest way to do that is a plain class whose methods are the middleware functions, so they share self:

class MiddlewareContainer:
    def __init__(self) -> None:
        self.call_count: int = 0  # ①

    async def call_counter_middleware(self, context, call_next) -> None:
        self.call_count += 1  # ②
        await call_next()

    async def result_enhancer_middleware(self, context, call_next) -> None:  # ③
        await call_next()
        if context.result:
            context.result = f"[Call #{self.call_count}] {context.result}"  # ④

① call_count lives on the instance, not inside either middleware function, so both methods can reach it through self. ② The counter middleware's only job is to increment that shared field before the rest of the chain runs. ③ The enhancer is a separate bound method on the same MiddlewareContainer instance, so it shares self with the counter. ④ It reads the counter's current value through self.call_count, which only works because both methods are bound to the same object.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/06-shared-state-middleware.py shows the runnable form.

Registering both bound methods from the same instance, middleware=[container.call_counter_middleware, container.result_enhancer_middleware], gives you two middleware that read and write the same self.call_count without any extra plumbing. Order matters here the same way it does anywhere in a chain: the counter has to run first so the enhancer sees an up-to-date count.

C# gets the same effect from a captured closure variable or a shared dictionary declared outside both middleware functions, since C# middleware is just delegates rather than bound methods on a shared object. There is also a narrower tool for the input-only case: the Use(sharedFunc: ...) overload lets one middleware serve both streaming and non-streaming runs without blocking the stream, at the cost of not being able to intercept or override the output, which makes it a good fit for pure inspection middleware that never needs context.result.

Where a runtime value actually lives

Middleware often needs to pass a value downstream: a tenant ID, a request ID, or something a tool needs but that should not leak into every tool's signature. Both languages give you three surfaces for this, and picking the wrong one either does not work or leaks data somewhere it should not go.

Surface	What it is for	Where you read it
Per-run metadata (AgentRunOptions.AdditionalProperties / function_invocation_kwargs=)	Values scoped to one run, forwarded into tool calls	Function middleware, tools
Session state (AgentSession.StateBag / session= + ctx.session)	Values that persist across runs in the same conversation	Any middleware or tool with access to the session
Function-invocation arguments (FunctionInvocationContext.Arguments / .kwargs)	Values scoped to a single tool call, inspected or rewritten in flight	Function invocation middleware only

That table maps each surface to its scope and its readers. In short:

• Surface: per-run metadata (AgentRunOptions.AdditionalProperties / function_invocation_kwargs=). What it's for: values scoped to one run, forwarded into tool calls. Where you read it: function middleware and tools.
• Surface: session state (AgentSession.StateBag / session= and ctx.session). What it's for: values that persist across runs in the same conversation. Where you read it: any middleware or tool with access to the session.
• Surface: function-invocation arguments (FunctionInvocationContext.Arguments / .kwargs). What it's for: values scoped to a single tool call, inspected or rewritten in flight. Where you read it: function invocation middleware only.

Use the narrowest one that fits. A tenant ID that changes per request belongs in run-level metadata, not session state, because session state outlives the run it was set in. A finding a tool needs to remember across the whole conversation belongs in session state, not run metadata, because run metadata disappears the moment the run finishes.

async def inject_function_kwargs(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    context.kwargs.setdefault("tenant", "contoso")
    await call_next()

That is function middleware writing into context.kwargs, which the tool then reads through its own injected FunctionInvocationContext parameter. C#'s equivalent reads AIAgent.CurrentRunContext?.RunOptions for the per-run bucket and writes straight into context.Arguments for the per-call bucket:

async ValueTask<object?> InjectRunContext(
    AIAgent agent,
    FunctionInvocationContext context,
    Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>> next,
    CancellationToken cancellationToken)
{
    if (AIAgent.CurrentRunContext?.RunOptions?.AdditionalProperties is { } props  // ①
        && props.TryGetValue("tenant", out var tenant))  // ②
    {
        context.Arguments["tenant"] = tenant;  // ③
    }
    return await next(context, cancellationToken);  // ④
}

① AIAgent.CurrentRunContext?.RunOptions?.AdditionalProperties is the per-run metadata bucket set when the run started. ② The middleware looks for a specific tenant key in that bucket, using TryGetValue so a missing key does not throw. ③ When found, it copies the value into context.Arguments, the per-call bucket the invoked function actually reads. ④ The chain always continues, so a request with no tenant set still reaches the function, just without that argument populated.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/07-inject-run-context.cs shows the runnable form.

Treat function_invocation_kwargs in Python as the replacement for the older pattern of passing arbitrary **kwargs straight into agent.run(). New tools should declare an explicit FunctionInvocationContext-annotated parameter and read from ctx.kwargs rather than accepting loose keyword arguments, since unexpected runtime kwargs get rejected outright.

Chat middleware: the layer closest to the model

Chat middleware is the one worth calling out separately, because it does not behave like the other two. Agent middleware wraps one call to agent.run(). Function middleware wraps one tool invocation. Chat middleware wraps every individual call to the model, and a single agent.run() can trigger several of those if the model calls a tool, gets a result, and calls the model again to finish answering.

async def logging_chat_middleware(
    context: ChatContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    print(f"[Chat] Sending {len(context.messages)} messages to AI")
    await call_next()
    print("[Chat] AI response received")

Register this on an agent that calls a tool once in one run, and the log line fires twice, once for the initial call, once after the tool result comes back, not once per agent.run() call. The sequence below traces a full round trip: the first model call requests a tool, the tool runs, and a second model call finishes the answer, with chat middleware wrapping both model calls.

That makes chat middleware the right place for anything that needs to see the model's actual conversation state turn by turn: token counting per call, prompt logging, or a modification that has to apply to every round trip rather than just the first one.

C# registers chat client middleware on the IChatClient itself, before it is handed to ChatClientAgent, rather than on the agent:

var middlewareEnabledChatClient = chatClient
    .AsBuilder()
        .Use(getResponseFunc: LoggingChatMiddleware, getStreamingResponseFunc: null)
    .Build();

var agent = new ChatClientAgent(middlewareEnabledChatClient, instructions: "You are a helpful assistant.");

Everything downstream of that IChatClient, the agent, its tool-calling loop, and any context providers, routes every model call through this middleware chain.

Exception handling: turning a crash into a graceful reply

Middleware is also the natural place to catch a tool that fails, because it sits between the failure and whatever the model or caller sees next. Function middleware wraps call_next() in a try/except and rewrites context.result instead of letting the exception propagate:

async def exception_handling_middleware(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    try:
        await call_next()
    except TimeoutError as e:
        context.result = (
            "Request Timeout: the data service is taking longer than expected. "
            "Respond with: 'Sorry for the inconvenience, please try again later.'"
        )

The tool itself raises TimeoutError however it normally would; the middleware never touches the tool's code. The model sees context.result as if the function had returned it directly, so a flaky downstream API degrades into a sensible reply instead of an unhandled crash reaching the caller. C# uses the same try-catch-and-return pattern at the agent-run level instead, wrapping innerAgent.RunAsync and returning a fallback AgentResponse on TimeoutException or any other caught exception.

Wiring a logging-and-validation pair onto a market-research agent

Take a market-research agent: it gathers information on a topic, tracks a plan, and produces a report. It should not just refuse to answer certain questions; it should refuse before spending a model call on them, and it should leave a trail of what it decided. That is two middleware: a validation guard that terminates early, and a logging middleware that records every run regardless of outcome.

A market-research agent has a specific version of the generic "blocked words" example every doc page shows: a request that asks it to act on insider information is not just off-topic; it is the one class of query this agent should never answer, no matter how it is phrased.

from agent_framework import (
    Agent,
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    Message,
    MiddlewareTermination,
    OpenAIChatClient,
)

class InsiderInfoGuardMiddleware(AgentMiddleware):
    """Blocks requests asking the agent to act on nonpublic information."""

    BLOCKED_TERMS = ("insider information", "material nonpublic", "mnpi")  # ①

    async def process(self, context: AgentContext, call_next) -> None:
        last_message = context.messages[-1] if context.messages else None
        query = last_message.text.lower() if last_message and last_message.text else ""

        for term in self.BLOCKED_TERMS:  # ②
            if term in query:
                context.result = AgentResponse(
                    messages=[Message(
                        role="assistant",
                        contents=["I can't research or act on nonpublic information. Ask about public filings, "
                                  "market data, or published analysis instead."],
                    )]
                )
                raise MiddlewareTermination(result=context.result)  # ③

        await call_next()


class RunLoggingMiddleware(AgentMiddleware):
    """Logs every run's outcome, including ones the guard middleware blocks."""

    async def process(self, context: AgentContext, call_next) -> None:
        print(f"[MarketResearcher] run starting, {len(context.messages)} messages in")  # ④
        await call_next()  # ⑤
        status = "blocked" if context.metadata.get("blocked") else "completed"  # ⑥
        print(f"[MarketResearcher] run {status}")


agent = Agent(
    client=OpenAIChatClient(),
    name="MarketResearcher",
    instructions="You are a market research analyst who gives clear, concise answers.",
    tools=[look_up_topic],
    middleware=[RunLoggingMiddleware(), InsiderInfoGuardMiddleware()],  # ⑦
    context_providers=[
        history,
        ResearchMemoryProvider("quantum computing funding trends", findings_store),
        compaction,
    ],
    require_per_service_call_history_persistence=True,
)

① The blocked-term policy is declared once as class data, not scattered through the check logic below. ② The guard loops over that policy list against the lowercased last message text. ③ On a match, it raises MiddlewareTermination, so look_up_topic never runs and no model call happens. ④ RunLoggingMiddleware prints before call_next(), so a starting line is recorded even when the guard is about to block the run. ⑤ It always calls call_next(), which is what lets the guard's termination, or a normal completion, happen inside this wrapper. ⑥ Reading context.metadata after call_next() returns tells the logger which of the two outcomes actually happened. ⑦ RunLoggingMiddleware is listed first, so it wraps InsiderInfoGuardMiddleware, giving it a single place to log both a block and a completion.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/08-market-research-middleware.py shows the runnable form, including stand-ins for the pieces referenced here (look_up_topic, history, compaction, ResearchMemoryProvider, findings_store).

Order is the point here. RunLoggingMiddleware is registered first, so it wraps InsiderInfoGuardMiddleware, which means it logs both outcomes, a normal completion and a termination, from one place instead of needing its own copy of the blocked-term check. The guard raises MiddlewareTermination before look_up_topic ever gets a chance to run, so a blocked request never reaches the model, never spends a token, and never touches the ResearchMemoryProvider findings store.

C# wires the same pair through the builder, with the logging middleware registered outermost so it wraps the guard the same way:

async Task<AgentResponse> InsiderInfoGuard(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    var query = messages.LastOrDefault()?.Text?.ToLower() ?? "";
    string[] blocked = ["insider information", "material nonpublic", "mnpi"];  // ①

    foreach (var term in blocked)  // ②
    {
        if (query.Contains(term))
        {
            return new AgentResponse([new ChatMessage(ChatRole.Assistant,  // ③
                "I can't research or act on nonpublic information. Ask about public filings, "
                + "market data, or published analysis instead.")]);
        }
    }

    return await innerAgent.RunAsync(messages, session, options, cancellationToken);  // ④
}

AIAgent agent = agentChatClient
    .AsBuilder()
        .Use(runFunc: RunLoggingMiddleware, runStreamingFunc: null)  // ⑤
        .Use(runFunc: InsiderInfoGuard, runStreamingFunc: null)  // ⑥
    .BuildAIAgent(new ChatClientAgentOptions
    {
        Name = "MarketResearcher",
        ChatOptions = new() { Instructions = "You are a market research analyst who gives clear, concise answers." },
    });

① The blocked-term policy is declared as a plain array, the same data-not-scattered-checks shape as the Python version. ② The guard loops over that array against the lowercased last message. ③ Returning a new AgentResponse here without calling innerAgent.RunAsync is C#'s equivalent of Python's MiddlewareTermination: the guard never lets the model see the request. ④ When nothing matches, the guard falls through and awaits the real inner agent, exactly as call_next() does in Python. ⑤ RunLoggingMiddleware is registered first in the builder chain, making it the outermost wrapper. ⑥ InsiderInfoGuard is registered second, so it runs inside the logging wrapper the same way InsiderInfoGuardMiddleware does in Python.

Note: The full extracted listing at code/microsoft_agent_framework/part-6-middleware/listings/09-market-research-middleware.cs shows the runnable form, including a stand-in RunLoggingMiddleware definition that mirrors the Python version above.

Neither version needs a new context provider, a new session field, or any change to the memory or compaction pipeline the agent already has. Middleware sits entirely outside that stack, which is exactly why it is the right tool for a check like this one.

Do this today

• Add one agent-level logging middleware that prints a start and an end line for every run. Wire it onto an agent that calls a tool, and watch how the true number of model calls compares to what you assumed.
• Write one guard middleware that raises MiddlewareTermination on a blocked term, and confirm the tool it protects never fires when the guard trips.
• Pick the narrowest state surface for any value you pass downstream. Run metadata for one call, session state for the whole conversation, function-invocation arguments for one tool call. Do not reach for session state by default just because it is the most flexible option.
• Wrap one flaky tool call in exception-handling middleware that turns a caught exception into a graceful context.result instead of letting the crash reach the caller.
• If any of your agents stream, audit every override middleware you have written for a context.stream check before it touches context.result. An override that only works for the non-streaming path is a bug waiting for the first streaming caller.

The takeaway

Middleware gives you three places to intercept an agent: wrapping a whole run, wrapping one tool call, or wrapping one model call, and every one of them follows the same call-chain contract: mutate the context, call the next link, and do your after-work once it returns. You have walked agent-level versus run-level scoping and the order they nest in, how to terminate a run before it starts, how to override a result while respecting the streaming-versus-non-streaming split, how to share state across middleware instances, and how to route a runtime value to the narrowest surface that actually needs it. Then you wired a logging-and-guard pair onto a market-research agent that blocks a whole class of request before it costs a token.

Middleware controls what an agent is allowed to see and say. What it does not touch is what an agent is allowed to do: write a file, call a paid API, or take any action a caller might regret if the agent gets it wrong. That is a different kind of gate, one built on approval rather than interception, and it is the next layer of trust worth putting on any agent that has graduated past answering questions into taking action.

---