Microsoft Agent

Microsoft Agent framework with Aiceberg

This document explains how we monitor different event types across an Agentic workflow in Agents built using Microsoft Agent Framework.

Aiceberg Middlewares provides real-time safety monitoring for your Microsoft Agent Framework Agents. It tracks user inputs, LLM calls, and tool executions and memory to ensure safe and compliant agent behavior.

Simply add it to your Microsoft Agent Framework Agents with:

middleware=[ AicebergAgentMiddleware(), AicebergFunctionMiddleware(), AicebergChatMiddleware() ]

and get instant visibility into the safety of your conversational and reasoning workflows.

What is middleware in agent framework ?

Middleware in the Agent Framework provides a powerful way to intercept, modify, and enhance agent interactions at various stages of execution. You can use middleware to implement cross-cutting concerns such as logging, security validation, error handling, and result transformation without modifying your core agent or function logic.

Middleware in agent framework can be function-based or class-based. This guide uses class-based middleware (suitable for complex logic). The Agent Framework exposes 3 types of middleware to monitor different event types:

• Agent middleware

• Chat middleware

• Function middleware

Aiceberg as a middleware

Aiceberg middleware in the Microsoft Agent Framework (MAF) acts as an advanced, real-time control layer within the agent execution flow, enabling security, compliance, and observability features for generative and multi-agent AI systems. By utilizing MAF's native middleware interfaces—specifically the Agent Run Middleware, Chat Client Middleware, and Function Calling Middleware—Aiceberg ensures that every user prompt, internal Large Language Model (LLM) request, tool invocation, and final agent response can be inspected and checked for policy violations, risks like prompt injection, and sensitive data leakage before reaching the model or returning to the user. This structured integration makes Aiceberg the essential enterprise-grade security and audit layer for MAF agents and complex workflows.

The Microsoft Agent Framework (MAF) has its own distinct structure for middleware that uses a specific set of interfaces and hook points. The key to Aiceberg's adaptation is leveraging the MAF's well-defined middleware architecture to function as a unified, real-time control plane across the entire agent lifecycle.

what events do we monitor	Middleware class	Where
User to Agent Forward User to Agent backward	Agentmiddleware	Around Agent run
Agent to LLM Forward Agent to LLM backward	Chatmiddleware	Around each LLM call
Agent to Tool Forward Agent to Tool backward	Functionmiddleware	Around each Tool call
Agent to memory forward Agent to memory backward for short term memory	Agentmiddleware	Around Agent run

---

Events Aiceberg monitors

1. User-to-Agent (user_agt)

The AicebergAgentMiddleware class is designed to wrap the entire execution of an Agent, intercepting the flow at the highest level—the Agent Run. This class is crucial for enforcing high-level security policies and compliance before and after any complex agent reasoning or tool-use occurs. This layer directly handles the overall agent run, from the initial user prompt to the final response. This is the ideal place to monitor the conversation from the user's and the agent's full perspective.

• Forward (User to Agent): Aiceberg performs initial content safety checks on the user's entire prompt, looking for prompt injection or jailbreaking attempts before the Large Language Model (LLM) processes the request. The entire agent execution within the Microsoft Agent Framework (MAF) can be stopped immediately if a policy violation or security risk is detected by Aiceberg by setting the context.terminate flag as True.

• Backward (Agent to User): Aiceberg performs final PII/sensitive data redaction and compliance checks on the agent's complete final response before it is displayed to the user. This ensures data leakage is prevented at the last possible moment.

Example AicebergAgentMiddleware:

AicebergAgentMiddleware.py

class AicebergAgentMiddleware(AgentMiddleware):
    """Agent middleware that logs execution."""

    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        response = aiceberg_monitor(is_input=True, content=context.messages[-1].text, event_type="user_agt", flagged=True, time_delay_flag=True, user_id="User")
        event_id = response.json().get("event_id")

        if response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
            context.result = AgentRunResponse(
                messages=[
                    ChatMessage(
                        role=Role.ASSISTANT,
                        text="Aiceberg blocked the Agent due to policy violations.",
                    )
                ]
            )
            context.terminate = True
            return

        await next(context)

        print(f"Agent Response: {context.result.text}")

        response = aiceberg_monitor(is_input=False, content=context.result.text, event_type="agt_llm", link_event_id=event_id, flagged=False, time_delay_flag=True, user_id="User")

2. Agent-to-LLM (agt_llm)

The AicebergChatMiddleware is implemented as the Chat Client Middleware in the MAF, acting as a mandatory firewall positioned directly between the agent's logic and the LLM service. Its primary responsibility is to inspect the exact payload being sent to the LLM—including the entire conversation history, the agent's system instructions, and the available tools—and, conversely, inspect the raw output received from the model.

• Forward (Agent to LLM): Executed before the LLM is invoked. Aiceberg monitors the entire payload—the system prompt, conversation history, and tool definitions—ensuring the agent's context hasn't been corrupted or subtly altered by adversarial inputs (persistent prompt injection). If flagged, the middleware can stop execution by setting context.terminate = True.

• Backward (LLM to Agent): Aiceberg inspects the raw LLM output for policy violations or toxic/misaligned content. It can block and terminate the agent execution at this stage by overriding the response in the ChatContext and setting the terminate flag to True.

Example AicebergChatMiddleware:

AicebergChatMiddleware.py

class AicebergChatMiddleware(ChatMiddleware):
    """Chat middleware that logs AI interactions."""

    async def process(
        self,
        context: ChatContext,
        next: Callable[[ChatContext], Awaitable[None]],
    ) -> None:
        request = context.to_dict()

        result_request = [
            {
                "role": msg["role"]["value"],
                "contents": msg["contents"]
            }
            for msg in request["messages"]
            if msg.get("type") == "chat_message"
        ]

        chat_options = request.get("chat_options", {})
        chat_options_filtered = {"instructions": chat_options.get("instructions")}

        Available_tools = []
        if hasattr(context.chat_options, "tools") and context.chat_options.tools:
            Available_tools = [(x.name, x.description) for x in context.chat_options.tools]

        chat_options_filtered["available_tools"] = Available_tools

        result_request_dict = {
            "messages": result_request,
            "chat_options": chat_options_filtered
        }

        prompt_response = aiceberg_response = aiceberg_monitor(
            is_input=True,
            content=str(result_request_dict),
            event_type="agt_llm",
            time_delay_flag=True,
            user_id="Agent"
        )

        event_id = aiceberg_response.json().get("event_id")

        if prompt_response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
            context.result = ChatResponse(
                messages=[
                    ChatMessage(
                        role=Role.ASSISTANT,
                        text="Aiceberg blocked the Agent due to policy violations.",
                    )
                ]
            )
            context.terminate = True
            return

        await next(context)

        response = context.result.to_dict()
        result_response = [
            {
                "role": msg["role"]["value"],
                "contents": msg["contents"]
            }
            for msg in response["messages"]
            if msg.get("type") == "chat_message"
        ]

        response = aiceberg_monitor(
            is_input=False,
            content=str(result_response),
            event_type="agt_llm",
            link_event_id=event_id,
            time_delay_flag=True,
            user_id="Agent"
        )

        if response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
            context.result = ChatResponse(
                messages=[
                    ChatMessage(
                        role=Role.ASSISTANT,
                        text="Aiceberg blocked the Agent due to policy violations.",
                    )
                ]
            )
            context.terminate = True
            return

3. Agent-to-tool

The AicebergFunctionMiddleware implements the MAF's Function Middleware interface and secures the boundary between the AI agent and external tools. By intercepting the FunctionInvocationContext, Aiceberg inspects the specific tool name and exact arguments generated by the LLM.

• Forward (Agent to Tools): Aiceberg intercepts the function name and arguments and checks for safety signals. If blocked, setting context.terminate = True stops that tool call (localized short-circuit). The agent receives a custom context.result, allowing the agent to continue reasoning without actually calling the tool.

• Backward (Tools to Agent): Aiceberg inspects the tool execution result before it is fed back to the LLM and checks for sensitive data.

Example AicebergFunctionMiddleware:

AicebergFunctionMiddleware.py

class AicebergFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function execution."""

    async def process(
        self,
        context: FunctionInvocationContext,
        next: Callable[[FunctionInvocationContext], Awaitable[None]],
    ) -> None:
        aiceberg_tool_dict = {}
        aiceberg_tool_dict["tool_name"]= context.function.name
        aiceberg_tool_dict["tool_args"] = context.arguments.model_dump()

        prompt_response = aiceberg_monitor(
            is_input=True,
            content=json.dumps(aiceberg_tool_dict),
            event_type="agt_tool",
            time_delay_flag=True,
            user_id="Agent"
        )

        event_id = prompt_response.json().get("event_id")

        if prompt_response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
            print("FunctionMiddleware: Terminating early")
            context.result = "Aiceberg blocked the Agent due to policy violations."
            context.terminate = True
            return

        await next(context)

        print(f"Function Response: {context.result}")

        response = aiceberg_monitor(
            is_input=False,
            content=str(context.result),
            event_type="agt_tool",
            link_event_id=event_id,
            time_delay_flag=True,
            user_id="Agent"
        )

        if response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
            print("FunctionMiddleware: Terminating early")
            context.result = "Aiceberg blocked the Agent due to policy violations."
            context.terminate = True
            return

Can the tool call be blocked? In the Microsoft Agent Framework, setting context.terminate = true within FunctionMiddleware acts as a localized "short-circuit" for a specific tool invocation rather than a global shutdown for the agent session. By omitting the await next(context) call and setting this flag, you effectively block the execution pipeline for that tool invocation while supplying a custom context.result so the agent can continue.

4. Agent-to-memory

In MAF, short-term memory is maintained within the AgentThread object, which holds the entire history of the conversation for a specific user session. This thread object is accessible to middleware via the AgentRunContext (context.thread). Aiceberg can inspect this memory at the beginning of every turn.

Memory Access and Inspection:

• Accessing History: middleware can access the list of previous conversations, tool calls, and agent responses stored in the thread's message store (context.thread.message_store.list_messages()).

• State Monitoring: by examining the serialized thread state, Aiceberg can audit the entire historical context passed to the LLM for the next turn.

Monitoring the short-term memory is critical for mitigating risks. Upon detecting severe threats within memory, Aiceberg can terminate the agent execution by setting context.terminate = True and overriding the result with a block message.

Example memory-inspection middleware:

LoggingAgentMiddleware.py

class LoggingAgentMiddleware(AgentMiddleware):
    """Agent middleware that logs execution."""

    async def process(
        self,
        context: AgentRunContext,
        next: Callable[[AgentRunContext], Awaitable[None]],
    ) -> None:
        if context.thread and context.thread.message_store:
            thread_messages = await context.thread.message_store.list_messages()
            all_history_content = []

            for message in thread_messages:
                role = message.role.value
                content = message.text
                if content:
                    all_history_content.append(f"[{role}]: {content}")

            full_context_string = "\n---\n".join(all_history_content)

            if full_context_string:
                monitor_response = aiceberg_monitor(
                    is_input=True,
                    content=full_context_string,
                    event_type="agt_mem",
                    time_delay_flag=True,
                    user_id="Agent"
                )

                event_id = monitor_response.json().get("event_id")

                if monitor_response.json().get("event_result") in ["flagged", "blocked", "rejected"]:
                    context.result = ChatResponse(
                        messages=[
                            ChatMessage(
                                role=Role.ASSISTANT,
                                text="Aiceberg blocked the Agent because of a policy violation found within the full conversation history (Memory Context).",
                            )
                        ]
                    )
                    context.terminate = True
                    return

        await next(context)

5. Agent-to-Agent / Multi-Agent Interaction Patterns

• Structured Agentic Workflows / Orchestrations: Graph-based orchestrations route data between executors/nodes and support sequential, concurrent (parallel), or handoff patterns. These are used for deterministic business processes and long-running tasks. Learn more.

• Agent as a Tool: One agent can be exposed to another as a callable function, enabling orchestrator agents to invoke specialized agents. When using the Agent-as-a-Tool pattern, AicebergFunctionMiddleware can monitor information passed between agents similar to standard agent-to-tool monitoring. Example:

agent_as_tool_example.py

def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    return f"The weather in {location} is cloudy with a high of 15°C."

async def basic_example():
    # Create an agent using OpenAI ChatCompletion
    weather_agent = OpenAIChatClient().create_agent(
        name="WeatherAgent",
        description="An agent that answers questions about the weather.",
        instructions="You answer questions about the weather.",
        tools=get_weather
    )

    main_agent = OpenAIChatClient().create_agent(
        name="Frenchassistant",
        instructions="You are a helpful assistant who responds in French.",
        tools=weather_agent.as_tool(),
        middleware=[AicebergAgentMiddleware(), AicebergFunctionMiddleware(), AicebergChatMiddleware()]
    )

    result = await main_agent.run("whats the pressure in New York?")

asyncio.run(basic_example())

Note on observability limits: agentic workflows/orchestrations can be hard to intercept via middleware because orchestration often happens via internal state mutations or direct function calls inside nodes/edges rather than through a standardized message bus or control plane. Blocking actions typically requires adding guard nodes or conditional logic directly into the workflow architecture.

---

Quick setup

1

Install the ab_microsoft_agentframework package

pip install -e .

2

Add environment variables (.env)

Set the following environment variables:

AICEBERG_API_URL= ... DEFAULT_HEADERS={"Content-Type": "application/json"} OPENAI_API_KEY=sk-... PROFILE_ID=... OPENAI_CHAT_MODEL_ID = ...

3

Register the Aiceberg middleware

Example: a Microsoft agent configured with AicebergAgentMiddleware, AicebergFunctionMiddleware, and AicebergChatMiddleware. This example demonstrates a simple Weather/Atmospheric pressure Agent that uses two tools — get_weather and get_pressure.

weather_agent_example.py

from typing import Annotated
from pydantic import Field
from agent_framework.openai import OpenAIChatClient
import asyncio
from ab_microsoft_agentframework.aiceberg_monitor import *

def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    return f"The weather in {location} is cloudy with a high of 15°C."

def get_pressure(
    location: Annotated[str, Field(description="The location to get the pressure for.")],
) -> str:
    """Get the atmospheric pressure for a given location."""
    return f"The atmospheric pressure in {location} is 1013 hPa."

async def basic_example():
    # Create an agent using OpenAI ChatCompletion
    agent = OpenAIChatClient().create_agent(
        name="WeatherAssistant",
        instructions="You are a helpful assistant.",
        tools=[get_weather, get_pressure],
        middleware=[AicebergAgentMiddleware(), AicebergFunctionMiddleware(), AicebergChatMiddleware()]
    )

    thread = agent.get_new_thread()
    result = await agent.run("whats the pressure in New York?", thread=thread)

asyncio.run(basic_example())

Run the agent and you should see the respective events on the Aiceberg dashboard.

---

Appendix — Distinction between function-based and class-based middlewares

In the Microsoft Agent Framework, the distinction centers on complexity and state:

• Function-based middleware: implemented as a simple asynchronous callback (e.g., via a decorator). Ideal for stateless tasks like basic logging or quick validation.

• Class-based middleware: implemented by subclassing abstract base classes like FunctionMiddleware or AgentMiddleware and implementing a process() method. Preferred for reusable, stateful components that need to maintain internal data across calls (e.g., security layers tracking violations, performance monitors).

Class-based middleware provides a more structured, object-oriented pattern suitable for Aiceberg's monitoring and enforcement use cases.