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 elke callback wordt call_next aangeroepen om door te gaan met verwerken. call_next neemt de context niet als argument; middleware muteert het gedeelde contextobject rechtstreeks en wacht vervolgens op call_next().

Opmerking

Middlewarevolgorde met gemengde registratiebereiken:

• Middleware op agentniveau verpakt middleware op run-level.
• Voor agent-middleware [A1, A2] en middleware [R1, R2]uitvoeren, is de uitvoeringsvolgorde: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• Function/chat middleware volgt hetzelfde wrapping-principe bij tool/chat-call time.

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
• session: De huidige agentsessie, indien van toepassing
• options: Opties voor agentuitvoering voor deze aanroep
• stream: 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)
• kwargs: Verouderde runtime-sleutelwoordargumenten die zijn doorgegeven aan de agentuitvoeringsmethode
• client_kwargs: Clientspecifieke runtimewaarden voor downstream-chatclients
• function_invocation_kwargs: Runtime-waarden die worden doorgestuurd naar hulpprogramma's

De call_next callback gaat verder met de middleware-keten of voert de agent uit als dit de laatste middleware is.

Op basis van functies

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

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,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await call_next()
        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
• session: De huidige agentsessie, indien van toepassing
• metadata: Woordenlijst voor het opslaan van extra gegevens tussen middleware
• result: de retourwaarde van de functie (kan worden gewijzigd)
• kwargs: Argumenten voor runtime-trefwoorden die worden doorgestuurd naar de aanroep van het hulpprogramma

De call_next callback gaat door naar de volgende middleware of voert de werkelijke functie uit.

Op basis van functies

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

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

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
• stream: 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)
• kwargs: Aanvullende trefwoordargumenten doorgegeven aan de chatclient
• function_invocation_kwargs: Runtime-waarden voor alleen hulpprogramma's die worden doorgestuurd door de chatlaag

De call_next callback gaat door naar de volgende middleware of verzendt de aanvraag naar de AI-service.

Op basis van functies

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

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

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

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(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 door instellingen in te stellen context.result en op te halen MiddlewareTermination. Dit is handig voor beveiligingscontroles, snelheidsbeperking of validatiefouten.

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

Wat beëindiging betekent:

• Instellen context.result voordat u een MiddlewareTermination aangepast antwoord wilt retourneren
• Het verhogen MiddlewareTermination stopt de rest van de middlewareketen en slaat het normale uitvoeringspad over
• Dit patroon werkt voor agent-, functie- en chat-middleware

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

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