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 anropsbar för att fortsätta bearbetningen 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
• is_streaming: Booleskt värde som anger om svaret strömmas
• metadata: Ordlista för lagring av ytterligare data mellan mellanprogram
• result: Agentens svar (kan ändras)
• terminate: Flagga för att stoppa ytterligare bearbetning
• kwargs: Ytterligare nyckelordsargument som skickas till agentkörningsmetoden

next Anropsbar fortsätter mellanprogramkedjan eller kör agenten om det är det sista mellanprogrammet.

Funktionsbaserad

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

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

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

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,
        next: Callable[[AgentContext], Awaitable[None]],
    ) -> None:
        print("[Agent Class] Starting execution")
        await next(context)
        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
• metadata: Ordlista för lagring av ytterligare data mellan mellanprogram
• result: Funktionens returvärde (kan ändras)
• terminate: Flagga för att stoppa ytterligare bearbetning
• kwargs: Ytterligare nyckelordsargument som skickas till chattmetoden som anropade den här funktionen

next Anropsbar fortsätter till nästa mellanprogram eller kör den faktiska funktionen.

Funktionsbaserad

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

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

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

Klassbaserad

from agent_framework import FunctionMiddleware, FunctionInvocationContext

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

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

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
• is_streaming: 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)
• terminate: Flagga för att stoppa ytterligare bearbetning
• kwargs: Ytterligare nyckelordsargument som skickas till chattklienten

Anropsbar next fortsätter till nästa mellanprogram eller skickar begäran till AI-tjänsten.

Funktionsbaserad

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

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

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

Klassbaserad

from agent_framework import ChatMiddleware, ChatContext

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

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

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, next):
    print("Before agent execution")
    await next(context)
    print("After agent execution")

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

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

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 med .context.terminate Detta är användbart för säkerhetskontroller, hastighetsbegränsning eller valideringsfel.

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

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

Vad avslutning innebär:

• Ange context.terminate = True signaler om att bearbetningen ska stoppas
• Du kan ge ett anpassat resultat innan du avslutar för att ge användarna feedback
• Agentkörningen hoppas över helt när mellanprogrammet avslutas

Å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.is_streaming för att skilja mellan dessa scenarier och hantera resultat åsidosättningar på lämpligt sätt.

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

    # Execute the original agent logic
    await next(context)

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

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

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

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

Bakgrundssvar för agent