Mellanprogram för agent

Mellanprogram i Agent Framework är ett kraftfullt sätt att fånga upp, ändra och förbättra agentinteraktioner i olika skeden av körningen. Du kan använda mellanprogram för att implementera övergripande problem som loggning, säkerhetsverifiering, felhantering och resultattransformering utan att ändra kärnagenten eller funktionslogik.

Agent Framework kan anpassas med tre olika typer av mellanprogram:

1. Mellanprogram för agentkörning: Tillåter avlyssning av alla agentkörningar, så att indata och utdata kan inspekteras och/eller ändras efter behov.
2. Funktion som anropar mellanprogram: Tillåter avlyssning av alla funktionsanrop som körs av agenten, så att indata och utdata kan inspekteras och ändras efter behov.
3. IChatClient middleware: Tillåter avlyssning av anrop till en IChatClient implementering, där en agent använder IChatClient för slutsatsdragningsanrop, till exempel när du använder ChatClientAgent.

Alla typer av mellanprogram implementeras via ett funktionsåteranrop, och när flera mellanprogramsinstanser av samma typ registreras bildar de en kedja, där varje mellanprograminstans förväntas anropa nästa i kedjan via en angivet nextFunc.

Agentkörning och funktionsanrop av mellanprogramstyper kan registreras på en agent med hjälp av agentverktyget med ett befintligt agentobjekt.

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

Viktigt!

Helst både runFunc och runStreamingFunc bör tillhandahållas. När agenten endast tillhandahåller mellanprogram som inte direktuppspelas använder den för både strömnings- och icke-direktuppspelningsanrop. Direktuppspelning körs endast i icke-strömningsläge för att uppfylla förväntningarna på mellanprogram.

Anmärkning

Det finns ytterligare en överlagring, Use(sharedFunc: ...), som gör att du kan tillhandahålla samma mellanprogram för icke-direktuppspelning och strömning utan att blockera strömningen. Det delade mellanprogrammet kan dock inte fånga upp eller åsidosätta utdata. Den här överlagringen bör användas för scenarier där du bara behöver inspektera eller ändra indata innan den når agenten.

IChatClient mellanprogram kan registreras på en IChatClient innan den används med ett ChatClientAgent, med hjälp av chattklientverktygets mönster.

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

Varning

DefaultAzureCredential är praktiskt för utveckling men kräver noggrant övervägande i produktion. I produktion bör du överväga att använda en specifik autentiseringsuppgift (t.ex. ManagedIdentityCredential) för att undvika problem med svarstid, oavsiktlig avsökning av autentiseringsuppgifter och potentiella säkerhetsrisker från reservmekanismer.

IChatClient mellanprogram kan också registreras med hjälp av en fabriksmetod när du skapar en agent via någon av hjälpmetoderna på SDK-klienter.

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

Mellanprogram för agentkörning

Här är ett exempel på mellanprogram för agentkörning som kan inspektera och/eller ändra indata och utdata från agentkörningen.

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

Mellanprogram för agentkörningsuppspelning

Här är ett exempel på agent som kör strömmande mellanprogram som kan inspektera och/eller ändra indata och utdata från agentströmningskörningen.

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

Funktion som anropar mellanprogram

Anmärkning

Mellanprogram för funktionsanrop stöds just nu endast med en AIAgent som använder FunctionInvokingChatClient, till exempel ChatClientAgent.

Här är ett exempel på funktion som anropar mellanprogram, som kan inspektera och/eller ändra funktionen som anropas och resultatet från funktionsanropet.

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

Det går att avsluta funktionsanropsloopen med funktionen som anropar mellanprogram genom att ställa in den angivna FunctionInvocationContext.Terminate till true. Detta förhindrar att funktionsanropsloopen skickar en begäran till inferenstjänsten som innehåller funktionsanropsresultatet efter funktionsanropet. Om det fanns fler än en funktion tillgänglig för anrop under den här iterationen kan det också förhindra att återstående funktioner körs.

Varning

Om du avslutar funktionsanropsloopen kan det leda till att chatthistoriken lämnas i ett inkonsekvent tillstånd, till exempel med funktionsanropsinnehåll utan funktionsresultatinnehåll. Detta kan leda till att chatthistoriken blir oanvändbar för ytterligare körningar.

IChatClient-mellanprogram

Här är ett exempel på mellanprogram för chattklienten som kan inspektera och/eller ändra indata och utdata för begäran till den slutsatsdragningstjänst som chattklienten tillhandahåller.

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

Tips/Råd

Se .NET-exemplen för fullständiga körbara exempel.

Anmärkning

Mer information om IChatClient mellanprogram finns i Custom IChatClient middleware.

Agent Framework kan anpassas med tre olika typer av mellanprogram:

1. Agentmellanprogram: Fångar upp körning av agentkörning så att du kan inspektera och ändra indata, utdata och kontrollflöde.
2. Funktionsmellanprogram: Avlyssnar funktionsanrop (verktyg) som görs under agentkörningen, vilket aktiverar validering av indata, resultattransformering och körningskontroll.
3. Chattmellanprogram: Fångar upp underliggande chattförfrågningar som skickas till AI-modeller, vilket ger åtkomst till rådata, alternativ och svar.

Alla typer stöder både funktionsbaserade och klassbaserade implementeringar. När flera mellanprogram av samma typ registreras bildar de en kedja där var och en anropar återanropet för att fortsätta bearbetningen call_next . call_next tar inte kontexten som ett argument. middleware muterar det delade kontextobjektet direkt och väntar sedan på call_next().

Anmärkning

Mellanprogramsordning med blandade registreringsomfattningar:

• Mellanprogram på agentnivå omsluter mellanprogram på körningsnivå.
• För mellanprogram [A1, A2] för agent och kör mellanprogram [R1, R2]är körningsordningen: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• Funktions-/chattmellanprogram följer samma omslutningsprincip vid verktygs-/chattsamtalstid.

Mellanprogram för agent

Agentens mellanprogram fångar upp och ändrar körningen av agentkörningen. Den använder som AgentContext innehåller:

• agent: Agenten som anropas
• messages: Lista över chattmeddelanden i konversationen
• session: Den aktuella agentsessionen, om någon
• options: Alternativ för agentkörning för det här anropet
• stream: Booleskt värde som anger om svaret strömmas
• metadata: Ordlista för lagring av ytterligare data mellan mellanprogram
• result: Agentens svar (kan ändras)
• kwargs: Argument för äldre körningsnyckelord som skickas till agentkörningsmetoden
• client_kwargs: Klientspecifika körningsvärden för underordnade chattklienter
• function_invocation_kwargs: Körningsvärden som vidarebefordras till verktyg

Återanropet call_next fortsätter mellanprogramskedjan eller kör agenten om det är det sista mellanprogrammet.

Funktionsbaserad

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

Klassbaserad

Klassbaserad agentmellanprogram använder en process metod som har samma signatur och beteende som funktionsbaserat mellanprogram.

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

Funktionsmellanprogram

Funktionsmellanprogram fångar upp funktionsanrop inom agenter. Den använder som FunctionInvocationContext innehåller:

• function: Funktionen som anropas
• arguments: De verifierade argumenten för funktionen
• session: Den aktuella agentsessionen, om någon
• metadata: Ordlista för lagring av ytterligare data mellan mellanprogram
• result: Funktionens returvärde (kan ändras)
• kwargs: Körningsnyckelordsargument som vidarebefordras till verktygets anrop

Återanropet call_next fortsätter till nästa mellanprogram eller kör den faktiska funktionen.

Funktionsbaserad

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

Klassbaserad

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

Chattmellanprogram

Chattmellanprogram fångar upp chattförfrågningar som skickas till AI-modeller. Den använder som ChatContext innehåller:

• chat_client: Chattklienten som anropas
• messages: Lista över meddelanden som skickas till AI-tjänsten
• options: Alternativen för chattbegäran
• stream: Booleskt värde som anger om detta är ett direktuppspelningsanrop
• metadata: Ordlista för lagring av ytterligare data mellan mellanprogram
• result: Chattsvaret från AI:n (kan ändras)
• kwargs: Ytterligare nyckelordsargument som skickas till chattklienten
• function_invocation_kwargs: Körningsvärden endast för verktyg som vidarebefordras av chattlagret

Återanropet call_next fortsätter till nästa mellanprogram eller skickar begäran till AI-tjänsten.

Anmärkning

Chattmellanprogram körs i funktionsanropsloopen. Det innebär att den körs för varje modellanrop, inklusive anrop som skickar verktygsresultat tillbaka till modellen under en anropssekvens med flera svängar.

Funktionsbaserad

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

Klassbaserad

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

Mellanprogramsdekoratörer

Dekoratörer tillhandahåller explicit mellanprogramstypdeklaration utan att kräva typanteckningar. De är användbara när du inte använder typanteckningar eller vill förhindra typmatchningar:

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

Registrering av mellanprogram

Mellanprogram kan registreras på två nivåer med olika omfång och beteenden.

Agent-Level vs 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?")

Viktiga skillnader:

• Agentnivå: Beständiga för alla körningar, konfigurerade en gång när agenten skapades
• Körningsnivå: Tillämpas endast på specifika körningar, tillåter anpassning per begäran
• Körningsordning: Agentmellanprogram (yttersta) → Kör mellanprogram (innersta) → Agentkörning

Mellanprogramsavslut

Mellanprogram kan avsluta körningen tidigt genom att ange context.result och höja MiddlewareTermination. Detta är användbart för säkerhetskontroller, hastighetsbegränsning eller valideringsfel.

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

Vad avslutning innebär:

• Ange context.result innan du lyfter MiddlewareTermination om du vill returnera ett anpassat svar
• Höjningen MiddlewareTermination stoppar resten av mellanprogramskedjan och hoppar över den normala körningsvägen
• Det här mönstret fungerar för mellanprogram för agent, funktion och chatt

Åsidosättning av mellanprogramresultat

Mellanprogram kan åsidosätta resultat i både scenarier som inte strömmas och strömning, så att du kan ändra eller helt ersätta agentsvar.

Resultattypen i context.result beror på om agentanropet är direktuppspelning eller inte:

• Icke-direktuppspelning: context.result innehåller ett AgentResponse med det fullständiga svaret
• Direktuppspelning: context.result innehåller en asynkron generator som ger AgentResponseUpdate segment

Du kan använda context.stream för att skilja mellan dessa scenarier och hantera resultat åsidosättningar på lämpligt sätt.

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

Med den här metoden för mellanprogram kan du implementera avancerad svarstransformering, innehållsfiltrering, resultatförbättring och anpassning av direktuppspelning samtidigt som agentlogik hålls ren och fokuserad.

Kompletta exempel på mellanprogram

Klassbaserat mellanprogram

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

Funktionsbaserat mellanprogram

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

Dekoratörsbaserat mellanprogram

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

Nästa steg

Definiera mellanprogram