Agent Middleware

Middleware in Agent Framework biedt een krachtige manier om interacties tussen agents in verschillende fasen van de uitvoering te onderscheppen, wijzigen en verbeteren. U kunt middleware gebruiken om kruislingse problemen te implementeren, zoals logboekregistratie, beveiligingsvalidatie, foutafhandeling en resultaattransformatie zonder de kernagent of functielogica te wijzigen.

Agent Framework kan worden aangepast met behulp van drie verschillende typen middleware:

1. Agent Run middleware: hiermee staat u onderschepping van alle agentuitvoeringen toe, zodat invoer en uitvoer naar behoefte kunnen worden geïnspecteerd en/of gewijzigd.
2. Middleware voor het aanroepen van functies: hiermee staat u interceptie toe van alle functie-aanroepen die door de agent worden uitgevoerd, zodat invoer en uitvoer naar behoefte kunnen worden geïnspecteerd en gewijzigd.
3. IChatClient middleware: maakt het onderscheppen van aanroepen naar een IChatClient implementatie mogelijk, waarbij een agent wordt gebruikt voor deductieaanroepen, bijvoorbeeld bij gebruik IChatClientChatClientAgent.

Alle typen middleware worden geïmplementeerd via een functie-callback en wanneer meerdere middleware-exemplaren van hetzelfde type zijn geregistreerd, vormen ze een keten, waarbij elke middleware-instantie naar verwachting de volgende in de keten aanroept, via een opgegeven nextFunc.

Agentuitvoering en functie die middlewaretypen aanroept, kunnen worden geregistreerd op een agent met behulp van de agentbouwer met een bestaand agentobject.

var middlewareEnabledAgent = originalAgent
    .AsBuilder()
        .Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: CustomAgentRunStreamingMiddleware)
        .Use(CustomFunctionCallingMiddleware)
    .Build();

Belangrijk

Idealiter zowel runFunc als runStreamingFunc moet worden voorzien. Wanneer u alleen de niet-streaming-middleware opgeeft, gebruikt de agent deze voor zowel streaming- als niet-streaming-aanroepen. Streaming wordt alleen uitgevoerd in de niet-streamingmodus om te voldoen aan de middlewareverwachtingen.

Opmerking

Er is een extra overbelasting, Use(sharedFunc: ...)waarmee u dezelfde middleware kunt bieden voor niet-streaming en streaming zonder de streaming te blokkeren. De gedeelde middleware kan de uitvoer echter niet onderscheppen of overschrijven. Deze overbelasting moet worden gebruikt voor scenario's waarbij u de invoer alleen hoeft te inspecteren of wijzigen voordat de agent wordt bereikt.

IChatClient middleware kan worden geregistreerd op een IChatClient voordat deze wordt gebruikt met een ChatClientAgent, met behulp van het patroon van de opbouwfunctie voor chatclients.

var chatClient = new AzureOpenAIClient(new Uri("https://<myresource>.openai.azure.com"), new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsIChatClient();

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

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

Waarschuwing

DefaultAzureCredential is handig voor ontwikkeling, maar vereist zorgvuldige overwegingen in de productieomgeving. Overweeg in productie een specifieke referentie te gebruiken (bijvoorbeeld ManagedIdentityCredential) om latentieproblemen, onbedoelde referentieprobing en potentiële beveiligingsrisico's van terugvalmechanismen te voorkomen.

IChatClient middleware kan ook worden geregistreerd met behulp van een factory-methode bij het maken van een agent via een van de helpermethoden op SDK-clients.

var agent = new AzureOpenAIClient(new Uri(endpoint), new DefaultAzureCredential())
    .GetChatClient(deploymentName)
    .AsAIAgent("You are a helpful assistant.", clientFactory: (chatClient) => chatClient
        .AsBuilder()
            .Use(getResponseFunc: CustomChatClientMiddleware, getStreamingResponseFunc: null)
        .Build());

Agent Middleware uitvoeren

Hier volgt een voorbeeld van agent run middleware, die de invoer en uitvoer van de agentuitvoering kan inspecteren en/of wijzigen.

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;
}

Agent Streaming Middleware uitvoeren

Hier volgt een voorbeeld van agent die streaming-middleware uitvoert, die de invoer en uitvoer van de streaming-uitvoering van de agent kan inspecteren en/of wijzigen.

    async IAsyncEnumerable<AgentResponseUpdate> CustomAgentRunStreamingMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    [EnumeratorCancellation] CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    List<AgentResponseUpdate> updates = [];
    await foreach (var update in innerAgent.RunStreamingAsync(messages, session, options, cancellationToken))
    {
        updates.Add(update);
        yield return update;
    }

    Console.WriteLine(updates.ToAgentResponse().Messages.Count);
}

Middleware voor functie-aanroepen

Opmerking

Middleware voor het aanroepen van functies wordt momenteel alleen ondersteund met een AIAgent die FunctionInvokingChatClient gebruikt, bijvoorbeeld ChatClientAgent.

Hier volgt een voorbeeld van de functie die middleware aanroept, die de aangeroepen functie kan inspecteren en/of wijzigen en het resultaat van de functieaanroep.

async ValueTask<object?> CustomFunctionCallingMiddleware(
    AIAgent agent,
    FunctionInvocationContext context,
    Func<FunctionInvocationContext, CancellationToken, ValueTask<object?>> next,
    CancellationToken cancellationToken)
{
    Console.WriteLine($"Function Name: {context!.Function.Name}");
    var result = await next(context, cancellationToken);
    Console.WriteLine($"Function Call Result: {result}");

    return result;
}

Het is mogelijk om de functie-aanroeplus te beëindigen met de functie die middleware aanroept door de opgegeven FunctionInvocationContext.Terminate waarde in te stellen op true. Hiermee voorkomt u dat de functieaanroepende lus een aanvraag verzendt naar de deductieservice met de resultaten van de functieaanroep na het aanroepen van de functie. Als er meer dan één functie beschikbaar was voor aanroepen tijdens deze iteratie, kan het ook voorkomen dat resterende functies worden uitgevoerd.

Waarschuwing

Als u de functieaanroeplus beëindigt, kan dit ertoe leiden dat uw chatgeschiedenis in een inconsistente status blijft, bijvoorbeeld met inhoud van functieaanroepen zonder inhoud van functieresultaten. Dit kan ertoe leiden dat de chatgeschiedenis onbruikbaar is voor verdere uitvoeringen.

IChatClient-middleware

Hier volgt een voorbeeld van chatclient-middleware, die de invoer en uitvoer voor de aanvraag kan inspecteren en/of wijzigen in de deductieservice die de chatclient levert.

async Task<ChatResponse> CustomChatClientMiddleware(
    IEnumerable<ChatMessage> messages,
    ChatOptions? options,
    IChatClient innerChatClient,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerChatClient.GetResponseAsync(messages, options, cancellationToken);
    Console.WriteLine(response.Messages.Count);

    return response;
}

Aanbeveling

Zie de .NET-voorbeelden voor volledige runnable voorbeelden.

Opmerking

Zie IChatClient voor meer informatie over middleware.

Agent Framework kan worden aangepast met behulp van drie verschillende typen middleware:

1. Agent-middleware: onderschept de uitvoering van de agent, zodat u invoer, uitvoer en controlestroom kunt inspecteren en wijzigen.
2. Functie-middleware: snijpuntfunctie (tool) aanroepen die zijn uitgevoerd tijdens het uitvoeren van de agent, het inschakelen van invoervalidatie, resultaattransformatie en uitvoeringsbeheer.
3. Chat-middleware: onderschept de onderliggende chataanvragen die naar AI-modellen worden verzonden, en biedt toegang tot de onbewerkte berichten, opties en antwoorden.

Alle typen ondersteunen implementaties op basis van functies en klassen. Wanneer meerdere middleware van hetzelfde type zijn geregistreerd, vormen ze een keten waarin elk de next aanroepbare aanroepbare aanroept om door te gaan met verwerken.

Agent Middleware

Agent-middleware onderschept en wijzigt de uitvoering van de agent. Hierbij wordt het AgentContext volgende gebruikt:

• agent: De agent die wordt aangeroepen
• messages: Lijst met chatberichten in het gesprek
• is_streaming: Booleaanse waarde die aangeeft of het antwoord streaming is
• metadata: Woordenlijst voor het opslaan van extra gegevens tussen middleware
• result: het antwoord van de agent (kan worden gewijzigd)
• terminate: Vlag om verdere verwerking te stoppen
• kwargs: Aanvullende trefwoordargumenten die zijn doorgegeven aan de agentuitvoeringsmethode

Het next aanroepbare vervolgt de middlewareketen of voert de agent uit als dit de laatste middleware is.

Op basis van functies

async def logging_agent_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], Awaitable[None]],
) -> None:
    """Agent middleware that logs execution timing."""
    # Pre-processing: Log before agent execution
    print("[Agent] Starting execution")

    # Continue to next middleware or agent execution
    await next(context)

    # Post-processing: Log after agent execution
    print("[Agent] Execution completed")

Op basis van klasse

Op klassen gebaseerde agent-middleware maakt gebruik van een process methode die dezelfde handtekening en hetzelfde gedrag heeft als op functies gebaseerde middleware.

from agent_framework import AgentMiddleware, AgentContext

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

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

Functie-middleware

Functie-middleware onderschept functie-aanroepen binnen agents. Hierbij wordt het FunctionInvocationContext volgende gebruikt:

• function: De functie die wordt aangeroepen
• arguments: De gevalideerde argumenten voor de functie
• metadata: Woordenlijst voor het opslaan van extra gegevens tussen middleware
• result: de retourwaarde van de functie (kan worden gewijzigd)
• terminate: Vlag om verdere verwerking te stoppen
• kwargs: Aanvullende trefwoordargumenten die zijn doorgegeven aan de chatmethode die deze functie heeft aangeroepen

De next aanroepbare blijft doorgaan naar de volgende middleware of voert de werkelijke functie uit.

Op basis van functies

async def logging_function_middleware(
    context: FunctionInvocationContext,
    next: Callable[[FunctionInvocationContext], Awaitable[None]],
) -> None:
    """Function middleware that logs function execution."""
    # Pre-processing: Log before function execution
    print(f"[Function] Calling {context.function.name}")

    # Continue to next middleware or function execution
    await next(context)

    # Post-processing: Log after function execution
    print(f"[Function] {context.function.name} completed")

Op basis van klasse

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

    async def process(
        self,
        context: FunctionInvocationContext,
        next: Callable[[FunctionInvocationContext], Awaitable[None]],
    ) -> None:
        print(f"[Function Class] Calling {context.function.name}")
        await next(context)
        print(f"[Function Class] {context.function.name} completed")

Chat-middleware

Chat-middleware onderschept chataanvragen die naar AI-modellen worden verzonden. Hierbij wordt het ChatContext volgende gebruikt:

• chat_client: De chatclient die wordt aangeroepen
• messages: Lijst met berichten die naar de AI-service worden verzonden
• options: De opties voor de chataanvraag
• is_streaming: Booleaanse waarde die aangeeft of dit een streamingaanroep is
• metadata: Woordenlijst voor het opslaan van extra gegevens tussen middleware
• result: Het chatantwoord van de AI (kan worden gewijzigd)
• terminate: Vlag om verdere verwerking te stoppen
• kwargs: Aanvullende trefwoordargumenten doorgegeven aan de chatclient

De next aanroepbare blijft naar de volgende middleware of verzendt de aanvraag naar de AI-service.

Op basis van functies

async def logging_chat_middleware(
    context: ChatContext,
    next: Callable[[ChatContext], Awaitable[None]],
) -> None:
    """Chat middleware that logs AI interactions."""
    # Pre-processing: Log before AI call
    print(f"[Chat] Sending {len(context.messages)} messages to AI")

    # Continue to next middleware or AI service
    await next(context)

    # Post-processing: Log after AI response
    print("[Chat] AI response received")

Op basis van klasse

from agent_framework import ChatMiddleware, ChatContext

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

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

Middleware Decorators

Decorators bieden expliciete middlewaretypedeclaratie zonder typeaantekeningen. Ze zijn handig als u geen typeaantekeningen gebruikt of wilt voorkomen dat type niet overeenkomt:

from agent_framework import agent_middleware, function_middleware, chat_middleware

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

@function_middleware
async def simple_function_middleware(context, next):
    print(f"Calling function: {context.function.name}")
    await next(context)
    print("Function call completed")

@chat_middleware
async def simple_chat_middleware(context, next):
    print(f"Processing {len(context.messages)} chat messages")
    await next(context)
    print("Chat processing completed")

Middlewareregistratie

Middleware kan worden geregistreerd op twee niveaus met verschillende bereiken en gedrag.

Agent-Level versus Run-Level Middleware

from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential

# Agent-level middleware: Applied to ALL runs of the agent
async with AzureAIAgentClient(async_credential=credential).as_agent(
    name="WeatherAgent",
    instructions="You are a helpful weather assistant.",
    tools=get_weather,
    middleware=[
        SecurityAgentMiddleware(),  # Applies to all runs
        TimingFunctionMiddleware(),  # Applies to all runs
    ],
) as agent:

    # This run uses agent-level middleware only
    result1 = await agent.run("What's the weather in Seattle?")

    # This run uses agent-level + run-level middleware
    result2 = await agent.run(
        "What's the weather in Portland?",
        middleware=[  # Run-level middleware (this run only)
            logging_chat_middleware,
        ]
    )

    # This run uses agent-level middleware only (no run-level)
    result3 = await agent.run("What's the weather in Vancouver?")

Belangrijkste verschillen:

• Agentniveau: Persistent voor alle uitvoeringen, eenmaal geconfigureerd bij het maken van de agent
• Uitvoeringsniveau: Alleen toegepast op specifieke uitvoeringen, maakt aanpassing per aanvraag mogelijk
• Uitvoeringsvolgorde: Agent middleware (buitenste) → Middleware uitvoeren (binnenste) → Agent-uitvoering

Beëindiging van middleware

Middleware kan de uitvoering vroegtijdig beëindigen met behulp van context.terminate. Dit is handig voor beveiligingscontroles, snelheidsbeperking of validatiefouten.

async def blocking_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], Awaitable[None]],
) -> None:
    """Middleware that blocks execution based on conditions."""
    # Check for blocked content
    last_message = context.messages[-1] if context.messages else None
    if last_message and last_message.text:
        if "blocked" in last_message.text.lower():
            print("Request blocked by middleware")
            context.terminate = True
            return

    # If no issues, continue normally
    await next(context)

Wat beëindiging betekent:

• Signalen instellen context.terminate = True die moeten worden verwerkt, moeten stoppen
• U kunt een aangepast resultaat opgeven voordat u afsluit om gebruikers feedback te geven
• De uitvoering van de agent wordt volledig overgeslagen wanneer middleware wordt beëindigd

Overschrijven van middlewareresultaat

Middleware kan resultaten overschrijven in scenario's die niet worden gestreamd en gestreamd, zodat u agentreacties kunt wijzigen of volledig kunt vervangen.

Het resultaattype is context.result afhankelijk van of de aanroep van de agent streaming of niet-streaming is:

• Niet-streaming: context.result bevat een AgentResponse met het volledige antwoord
• Streaming: context.result bevat een asynchrone generator die segmenten oplevert AgentResponseUpdate

U kunt dit gebruiken context.is_streaming om onderscheid te maken tussen deze scenario's en om resultaatoverschrijvingen op de juiste manier af te handelen.

async def weather_override_middleware(
    context: AgentContext,
    next: Callable[[AgentContext], Awaitable[None]]
) -> None:
    """Middleware that overrides weather results for both streaming and non-streaming."""

    # Execute the original agent logic
    await next(context)

    # Override results if present
    if context.result is not None:
        custom_message_parts = [
            "Weather Override: ",
            "Perfect weather everywhere today! ",
            "22°C with gentle breezes. ",
            "Great day for outdoor activities!"
        ]

        if context.is_streaming:
            # Streaming override
            async def override_stream() -> AsyncIterable[AgentResponseUpdate]:
                for chunk in custom_message_parts:
                    yield AgentResponseUpdate(contents=[Content.from_text(text=chunk)])

            context.result = override_stream()
        else:
            # Non-streaming override
            custom_message = "".join(custom_message_parts)
            context.result = AgentResponse(
                messages=[Message(role="assistant", contents=[custom_message])]
            )

Met deze middleware-benadering kunt u geavanceerde reactietransformatie, inhoudsfiltering, verbetering van resultaten en streaming-aanpassing implementeren, terwijl uw agentlogica schoon en gericht blijft.

Volledige middlewarevoorbeelden

Op klassen gebaseerde middleware

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Op functies gebaseerde middleware

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Middleware op basis van decorator

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


class LoggingFunctionMiddleware(FunctionMiddleware):
    """Function middleware that logs function calls."""

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Volgende stappen

Achtergrondantwoorden van agent