Intergiciel de l’agent

L’intergiciel dans Agent Framework offre un moyen puissant d’intercepter, de modifier et d’améliorer les interactions de l’agent à différentes étapes de l’exécution. Vous pouvez utiliser l’intergiciel pour implémenter des préoccupations croisées telles que la journalisation, la validation de la sécurité, la gestion des erreurs et la transformation des résultats sans modifier votre logique principale d’agent ou de fonction.

Agent Framework peut être personnalisé à l’aide de trois types différents d’intergiciels :

1. Intergiciel d’exécution de l’agent : autorise l’interception de toutes les exécutions de l’agent, afin que l’entrée et la sortie puissent être inspectées et/ou modifiées selon les besoins.
2. Intergiciel d’appel de fonction : autorise l’interception de tous les appels de fonction exécutés par l’agent, afin que l’entrée et la sortie puissent être inspectées et modifiées selon les besoins.
3. IChatClient middleware : autorise l’interception des appels à une IChatClient implémentation, où un agent utilise IChatClient pour les appels d’inférence, par exemple lors de l’utilisation ChatClientAgent.

Tous les types d’intergiciels sont implémentés via un rappel de fonction, et lorsque plusieurs instances d’intergiciels du même type sont inscrites, elles forment une chaîne, où chaque instance d’intergiciel est censée appeler la suivante dans la chaîne, via une chaîne fournie nextFunc.

Les types d’intergiciels d’exécution et de fonction de l’agent peuvent être inscrits sur un agent à l’aide du générateur d’agents avec un objet agent existant.

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

Important

Idéalement, les deux runFunc et runStreamingFunc doivent être fournis. Lorsque vous fournissez simplement l’intergiciel non en streaming, l’agent l’utilisera pour les appels de diffusion en continu et de non-diffusion en continu. La diffusion en continu s’exécute uniquement en mode non streaming pour suffire aux attentes des intergiciels.

Note

Il existe une surcharge supplémentaire, Use(sharedFunc: ...)qui vous permet de fournir le même intergiciel pour le non-streaming et la diffusion en continu sans bloquer la diffusion en continu. Toutefois, le middleware partagé ne pourra pas intercepter ou remplacer la sortie. Cette surcharge doit être utilisée pour les scénarios où vous devez uniquement inspecter ou modifier l’entrée avant d’atteindre l’agent.

IChatClient l’intergiciel peut être inscrit sur un IChatClient intergiciel avant d’être utilisé avec un ChatClientAgentmodèle de générateur client de conversation.

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.");

Avertissement

DefaultAzureCredential est pratique pour le développement, mais nécessite une considération minutieuse en production. En production, envisagez d’utiliser des informations d’identification spécifiques (par exemple ManagedIdentityCredential) pour éviter les problèmes de latence, la détection involontaire des informations d’identification et les risques de sécurité potentiels liés aux mécanismes de secours.

IChatClient le middleware peut également être inscrit à l’aide d’une méthode de fabrique lors de la construction d’un agent via l’une des méthodes d’assistance des clients SDK.

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());

Middleware d’exécution de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de l’agent.

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

Intergiciel d’exécution de streaming de l’agent

Voici un exemple d’intergiciel d’exécution d’agent, qui peut inspecter et/ou modifier l’entrée et la sortie de l’exécution de streaming de l’agent.

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

Intergiciel appelant la fonction

Note

Le middleware d’appel de fonction est actuellement pris en charge uniquement avec un AIAgent qui utilise FunctionInvokingChatClient, par exemple ChatClientAgent.

Voici un exemple d’intergiciel appelant une fonction, qui peut inspecter et/ou modifier la fonction appelée et le résultat de l’appel de fonction.

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

Il est possible d’arrêter la boucle d’appel de fonction avec l’intergiciel d’appel de fonction en définissant la valeur true fournie FunctionInvocationContext.Terminate . Cela empêche la boucle d’appel de fonction d’émettre une requête au service d’inférence contenant les résultats de l’appel de fonction après l’appel de fonction. S’il existe plusieurs fonctions disponibles pour l’appel pendant cette itération, il peut également empêcher l’exécution des fonctions restantes.

Avertissement

La fin de la boucle d’appel de fonction peut entraîner la gauche de votre historique des conversations dans un état incohérent, par exemple, contenant du contenu d’appel de fonction sans contenu de résultat de fonction. Cela peut entraîner l’inutilisable de l’historique des conversations pour d’autres exécutions.

Intergiciel IChatClient

Voici un exemple d’intergiciel client de conversation, qui peut inspecter et/ou modifier l’entrée et la sortie de la demande au service d’inférence fourni par le client de conversation.

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

Conseil / Astuce

Consultez les exemples .NET pour obtenir des exemples exécutables complets.

Note

Pour plus d’informations sur IChatClient middleware, consultez l’intergiciel personnalisé IChatClient.

Agent Framework peut être personnalisé à l’aide de trois types différents d’intergiciels :

1. Intergiciel de l’agent : intercepte l’exécution de l’agent, ce qui vous permet d’inspecter et de modifier les entrées, les sorties et le flux de contrôle.
2. Intergiciel de fonction : intercepte les appels de fonction (outil) effectués pendant l’exécution de l’agent, en activant la validation d’entrée, la transformation des résultats et le contrôle d’exécution.
3. Intergiciel de conversation : intercepte les demandes de conversation sous-jacentes envoyées aux modèles IA, en fournissant l’accès aux messages bruts, aux options et aux réponses.

Tous les types prennent en charge les implémentations basées sur des fonctions et basées sur des classes. Lorsque plusieurs intergiciels du même type sont inscrits, ils forment une chaîne où chaque appel appelle le rappel pour continuer le call_next traitement. call_next ne prend pas le contexte comme argument ; middleware mute directement l’objet de contexte partagé, puis attend call_next().

Note

Ordre des intergiciels avec des étendues d’inscription mixtes :

• L’intergiciel au niveau de l’agent encapsule les intergiciels au niveau de l’exécution.
• Pour l’intergiciel [A1, A2] agent et l’intergiciel [R1, R2]d’exécution, l’ordre d’exécution est : A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• L’intergiciel de fonction/conversation suit le même principe d’habillage au moment de l’outil/de l’appel de conversation.

Intergiciel de l’agent

L’intergiciel agent intercepte et modifie l’exécution de l’exécution de l’agent. Il utilise les AgentContext éléments suivants :

• agent: l’agent appelé
• messages: liste des messages de conversation dans la conversation
• session: session de l’agent en cours, le cas échéant
• options: Options d’exécution de l’agent pour cet appel
• stream: booléen indiquant si la réponse est en streaming
• metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
• result: la réponse de l’agent (peut être modifiée)
• kwargs: arguments de mot clé runtime hérités passés à la méthode d’exécution de l’agent
• client_kwargs: valeurs d’exécution spécifiques au client pour les clients de conversation en aval
• function_invocation_kwargs: valeurs d’exécution qui seront transférées aux outils

Le call_next rappel poursuit la chaîne d’intergiciels ou exécute l’agent s’il s’agit du dernier middleware.

Basé sur des fonctions

async def inject_tool_runtime_defaults(
    context: AgentContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Agent middleware that sets tool-only runtime defaults."""
    print("[Agent] Starting execution")
    context.function_invocation_kwargs.setdefault("tenant", "contoso")
    context.function_invocation_kwargs.setdefault("request_source", "agent-middleware")

    await call_next()

    print("[Agent] Execution completed")

Basé sur des classes

L’intergiciel d’agent basé sur une classe utilise une process méthode qui a la même signature et le même comportement que l’intergiciel basé sur les fonctions.

from agent_framework import AgentMiddleware, AgentContext

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

    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")

Intergiciel de fonction

L’intergiciel de fonction intercepte les appels de fonction au sein des agents. Il utilise les FunctionInvocationContext éléments suivants :

• function: fonction appelée
• arguments: arguments validés pour la fonction
• session: session de l’agent en cours, le cas échéant
• metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
• result: valeur de retour de la fonction (peut être modifiée)
• kwargs: arguments de mot clé runtime qui seront transférés à l’appel de l’outil

Le call_next rappel passe au middleware suivant ou exécute la fonction réelle.

Basé sur des fonctions

async def inject_function_kwargs(
    context: FunctionInvocationContext,
    call_next: Callable[[], Awaitable[None]],
) -> None:
    """Function middleware that enriches tool runtime values."""
    context.kwargs.setdefault("tenant", "contoso")
    context.kwargs.setdefault("request_source", "function-middleware")

    await call_next()

Basé sur des classes

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

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

Intergiciel de conversation

L’intergiciel de conversation intercepte les demandes de conversation envoyées aux modèles IA. Il utilise les ChatContext éléments suivants :

• chat_client: le client de conversation appelé
• messages: liste des messages envoyés au service IA
• options: options de la demande de conversation
• stream: booléen indiquant s’il s’agit d’un appel de diffusion en continu
• metadata: Dictionnaire pour le stockage de données supplémentaires entre intergiciels
• result: Réponse de conversation de l’IA (peut être modifiée)
• kwargs: arguments de mot clé supplémentaires passés au client de conversation
• function_invocation_kwargs: valeurs d’exécution de l’outil uniquement qui seront transférées par la couche de conversation

Le call_next rappel continue à l’intergiciel suivant ou envoie la requête au service IA.

Basé sur des fonctions

async def logging_chat_middleware(
    context: ChatContext,
    call_next: Callable[[], 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 call_next()

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

Basé sur des classes

from agent_framework import ChatMiddleware, ChatContext

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

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

Décorateurs d’intergiciels

Les décorateurs fournissent une déclaration de type middleware explicite sans nécessiter d’annotations de type. Ils sont utiles lorsque vous n’utilisez pas d’annotations de type ou que vous souhaitez empêcher les incompatibilités de type :

from agent_framework import agent_middleware, function_middleware, chat_middleware

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

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

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

Inscription d’intergiciel

Les intergiciels peuvent être inscrits à deux niveaux avec des étendues et des comportements différents.

middleware Agent-Level vs Run-Level

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(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?")

Principales différences :

• Niveau de l’agent : persistant sur toutes les exécutions, configuré une fois lors de la création de l’agent
• Niveau d’exécution : appliqué uniquement aux exécutions spécifiques, autorise la personnalisation par requête
• Ordre d’exécution : Intergiciel agent (externe) → Exécuter l’intergiciel (le plus interne) → Exécution de l’agent

Arrêt du middleware

L’intergiciel peut arrêter l’exécution tôt en définissant context.result et en générant MiddlewareTermination. Cela est utile pour les vérifications de sécurité, la limitation du débit ou les échecs de validation.

from agent_framework import AgentContext, AgentResponse, Message, MiddlewareTermination

async def blocking_middleware(
    context: AgentContext,
    call_next: Callable[[], 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.result = AgentResponse(
                messages=[Message(role="assistant", text="This request was blocked by middleware.")]
            )
            raise MiddlewareTermination(result=context.result)

    # If no issues, continue normally
    await call_next()

Ce que signifie l’arrêt :

• Définir context.result avant de déclencher MiddlewareTermination si vous souhaitez retourner une réponse personnalisée
• Le déclenchement MiddlewareTermination arrête le reste de la chaîne d’intergiciels et ignore le chemin d’exécution normal
• Ce modèle fonctionne pour l’agent, la fonction et le middleware de conversation

Remplacement des résultats du middleware

Le middleware peut remplacer les résultats dans les scénarios de diffusion en continu et de diffusion en continu, ce qui vous permet de modifier ou de remplacer complètement les réponses de l’agent.

Le type context.result de résultat varie selon que l’appel de l’agent est en streaming ou non en streaming :

• Non-streaming : context.result contient une AgentResponse réponse complète
• Streaming : context.result contient un générateur asynchrone qui génère AgentResponseUpdate des blocs

Vous pouvez utiliser context.stream pour différencier ces scénarios et gérer les remplacements de résultats de manière appropriée.

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

    # Execute the original agent logic
    await call_next()

    # 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.stream:
            # 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])]
            )

Cette approche d’intergiciel vous permet d’implémenter une transformation de réponse sophistiquée, un filtrage de contenu, une amélioration des résultats et une personnalisation du streaming tout en conservant la logique de votre agent propre et prioritaire.

Exemples complets d’intergiciels

Intergiciel basé sur des classes

# 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())

Intergiciel basé sur des fonctions

# 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())

Intergiciel basé sur le décorateur

# 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())

Prochaines étapes

Réponses en arrière-plan de l’agent