Building with Agent Framework - Day 1: Block, Transform, and Control Agent Behavior with Middleware

#47 | Add reusable validation, security, and logging without touching your core agent logic

I mentioned earlier that I’d build a demo-a-day for the next week after the launch of the Microsoft Agent Framework - because frameworks are a dime a dozen, but what matters is what you can build with them.

Introducing Microsoft Agent Framework (Semantic Kernel + AutoGen)

A demo a day keeps the FOMO at bay.

In the first demo today, I’ll explore a feature in Agent Framework that I think is extremely useful: Middleware. You can think of middleware as a way to intercept and modify the behavior of your agent at various stages of its operation. If you have built web applications - you have probably seen similar middleware patterns where the idea is to build reusable components (e.g., auth, logging, etc) that can intercept, modify or reject the requests that come to your server.

We’ll start with a simple agent with a tool that can tell the weather. Only in this case, there’s a super special location - one that no one must ever speak of - Atlantis. Other locations are fair game. Our agent must never speak of Atlantis and must warn users accordingly.

While this demo scenario is whimsical, later in this post I’ll show how the same patterns apply to critical real-world scenarios like blocking PII data, enforcing rate limits, and implementing security controls.

---

---

Building the Weather Agent

A basic agent with a tool.

Let’s start with a basic weather agent that has a simple function to get the current weather:

from typing import Annotated
from agent_framework import ChatAgent
from agent_framework.azure import AzureOpenAIChatClient
import os

def get_weather(
    location: Annotated[str, “The location to get the weather for.”],
) -> str:
    “”“Get the weather for a given location.”“”
    conditions = [”sunny”, “cloudy”, “rainy”, “stormy”]
    temperature = 53
    return f”The weather in {location} is {conditions[0]} with a high of {temperature}°C.”

# Create the agent
agent = ChatAgent(
    name=”WeatherAgent”,
    description=”A helpful agent that provides weather information”,
    instructions=”You are a weather assistant. Provide current weather information for any location.”,
    chat_client=AzureOpenAIChatClient(
        api_key=os.environ.get(”AZURE_OPENAI_API_KEY”, “”),
    ),
    tools=[get_weather],
)

This agent can answer questions like “What’s the weather in Paris?” The agent will call the get_weather function and return the results.

Adding Middleware to Block Forbidden Locations

Now comes the interesting part. What if we want to prevent anyone from asking about the weather in Atlantis? We could add checks inside the get_weather function, but that would mix business logic with validation logic. Instead, let’s use function middleware to intercept the function call before it executes:

from agent_framework import (
    FunctionInvocationContext,
    function_middleware,
)
from collections.abc import Awaitable, Callable

@function_middleware
async def atlantis_location_filter_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    “”“Function middleware that blocks weather requests for Atlantis.”“”
    # Check if location parameter is “atlantis”
    location = getattr(context.arguments, “location”, None)
    if location and location.lower() == “atlantis”:
        context.result = (
            “Blocked! Hold up right there!! Tell the user that “
            “’Atlantis is a special place, we must never ask about the weather there!!’”
        )
        context.terminate = True
        return

    await next(context)

Now add the middleware to the agent:

agent = ChatAgent(
    name=”WeatherAgent”,
    description=”A helpful agent that provides weather information”,
    instructions=”You are a weather assistant. Provide current weather information for any location.”,
    chat_client=AzureOpenAIChatClient(
        api_key=os.environ.get(”AZURE_OPENAI_API_KEY”, “”),
    ),
    tools=[get_weather],
    middleware=[atlantis_location_filter_middleware],  # Add middleware here
)

We use a middleware to detect the location argument for the weather tool and replace the result based on some condition.

What’s happening here?

1. The @function_middleware decorator marks this as function-level middleware

2. The middleware inspects context.arguments to check the location parameter

3. If it’s “Atlantis”, it sets context.result to override the function’s response

4. Setting context.terminate = True stops the pipeline - the actual function never executes

5. If it’s not Atlantis, await next(context) continues to the actual function

When someone asks “What’s the weather in Atlantis?”, they’ll get our special blocked message instead!

Bonus: Adding Security with Chat Middleware

Function middleware is great for intercepting tool calls, but what if we want to block requests before they even reach the LLM? That’s where chat middleware comes in:

from agent_framework import (
    ChatContext,
    ChatMessage,
    ChatResponse,
    Role,
    chat_middleware,
)

@chat_middleware
async def security_filter_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    “”“Chat middleware that blocks requests containing sensitive information.”“”
    blocked_terms = [”password”, “secret”, “api_key”, “token”]

    for message in context.messages:
        if message.text:
            message_lower = message.text.lower()
            for term in blocked_terms:
                if term in message_lower:
                    # Override the response without calling the LLM
                    context.result = ChatResponse(
                        messages=[
                            ChatMessage(
                                role=Role.ASSISTANT,
                                text=(
                                    “I cannot process requests containing sensitive information. “
                                    “Please rephrase your question without including passwords, secrets, “
                                    “or other sensitive data.”
                                ),
                            )
                        ]
                    )
                    return

    await next(context)

Add it to the agent’s middleware list:

agent = ChatAgent(
    # ... other configuration ...
    middleware=[security_filter_middleware, atlantis_location_filter_middleware],
)

This middleware runs before the LLM is called, saving you API costs by blocking inappropriate requests early!

With middleware we can detect and block sensitive content in agent messages.

Three Types of Middleware in Agent Framework

The framework supports three types of middleware, each intercepting at different stages:

1. Agent Middleware

Intercepts before and after the entire agent execution. Perfect for:

• Logging agent invocations

• Retry logic

• Performance monitoring

• Authentication/authorization

from agent_framework import AgentRunContext, agent_middleware

@agent_middleware
async def logging_middleware(
    context: AgentRunContext,
    next: Callable[[AgentRunContext], Awaitable[None]],
) -> None:
    print(f”[Agent] Starting: {context.agent.name}”)
    await next(context)
    print(f”[Agent] Completed: {context.result}”)

2. Function Middleware

Intercepts tool/function calls. Perfect for:

• Input validation

• Caching results

• Access control (like our Atlantis example!)

• Argument transformation

from agent_framework import FunctionInvocationContext, function_middleware

@function_middleware
async def validation_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    print(f”[Function] Calling: {context.function.name}”)
    await next(context)
    print(f”[Function] Result: {context.result}”)

3. Chat Middleware

Intercepts LLM requests. Perfect for:

• Content filtering

• Prompt injection prevention

• Token counting

• Message modification

from agent_framework import ChatContext, chat_middleware

@chat_middleware
async def token_counter_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    context.metadata[”input_messages”] = len(context.messages)
    await next(context)
    # Access response and count tokens

Implementation Styles

Agent Framework gives you flexibility in how you write middleware:

Style 1: Function-based with Decorators (Recommended)

Simple and clean - no type annotations needed:

@function_middleware
async def simple_middleware(context, next):
    await next(context)

Style 2: Function-based with Type Annotations

The framework detects middleware type from parameter types:

async def typed_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    await next(context)

Style 3: Class-based

For stateful middleware or complex logic:

from agent_framework import FunctionMiddleware

class CachingMiddleware(FunctionMiddleware):
    def __init__(self):
        self.cache = {}

    async def process(self, context: FunctionInvocationContext, next):
        cache_key = f”{context.function.name}:{context.arguments}”

        if cache_key in self.cache:
            context.result = self.cache[cache_key]
            context.terminate = True
            return

        await next(context)

        if context.result:
            self.cache[cache_key] = context.result

Key Middleware Patterns

Here are the essential patterns you’ll use when building middleware:

Pattern 1: Terminate Execution

Stop the pipeline without executing the underlying function/agent:

@function_middleware
async def rate_limit_middleware(context, next):
    if is_rate_limited(context.function.name):
        context.result = “Rate limit exceeded. Please try again later.”
        context.terminate = True  # Stop here, don’t execute the function
        return

    await next(context)

Pattern 2: Override Results

Execute normally but replace the result:

@function_middleware
async def cache_middleware(context, next):
    cache_key = f”{context.function.name}:{context.arguments}”

    # Check cache first
    if cache_key in cache:
        context.result = cache[cache_key]
        context.terminate = True
        return

    # Execute and cache result
    await next(context)
    cache[cache_key] = context.result

Pattern 3: Share Data Between Middleware

Use context.metadata to pass information through the pipeline:

@agent_middleware
async def timer_middleware(context, next):
    import time
    context.metadata[”start_time”] = time.time()
    await next(context)
    duration = time.time() - context.metadata[”start_time”]
    print(f”Execution took {duration:.2f}s”)

Pattern 4: Transform Before and After

Modify inputs before execution and outputs after:

@function_middleware
async def sanitize_middleware(context, next):
    # Transform input
    location = getattr(context.arguments, “location”, None)
    if location:
        context.arguments.location = location.strip().title()

    # Execute
    await next(context)

    # Transform output
    if context.result:
        context.result = context.result.upper()

Wrapping Up

Middleware in Agent Framework provides clean separation of concerns - your business logic stays in your functions while cross-cutting concerns like validation, logging, and security live in reusable middleware.

The Atlantis example may be whimsical, but the pattern is powerful for real-world scenarios:

• Blocking PII from being sent to LLMs

• Enforcing rate limits on expensive tool calls

• Adding audit trails for compliance

• Implementing caching layers for performance

Want to learn more? Check out the complete examples in the Agent Framework repository:

• Getting Started with Middleware

• Decorator-based Middleware

• Class-based Middleware

• Exception Handling with Middleware

• Override Results with Middleware

What’s Next?

Day 2? What if we try to build an agent that helps you migrate your existing agent code from AutoGen or SK to Agent Framework?

---

P.S - I wrote a book!

I wrote a book - Designing Multi-Agent Systems - which is in early access, with the full version ready on Nov 10. If this type of content is interesting to you, you might find the book useful. The book takes a from scratch approach - for example Chapter 4 - Building An Agent from Scratch covers the core concepts of middleware, how they are built and what they accomplish.

I wrote a book!

You can look at a preview of the book before you buy.

• Preview: https://multiagentbook.com/preview.pdf

• Early Access: https://buy.multiagentbook.com/