Middleware agenta

Middleware v rozhraní Agent Framework poskytuje účinný způsob, jak zachytit, upravit a vylepšit interakce agentů v různých fázích provádění. Middleware můžete použít k implementaci průřezových aspektů, jako je protokolování, ověřování zabezpečení, zpracování chyb a transformace výsledků beze změny základního agenta nebo logiky funkce.

Rozhraní Agent Framework je možné přizpůsobit pomocí tří různých typů middlewaru:

1. Spuštění middlewaru agenta: Umožňuje zachytit všechna spuštění agenta, aby bylo možné v případě potřeby zkontrolovat vstup a výstup nebo upravit.
2. Middleware volání funkce: Umožňuje zachytit všechna volání funkcí spuštěná agentem, aby bylo možné podle potřeby kontrolovat a upravovat vstup a výstup.
3. IChatClient middleware: Umožňuje zachycení volání do IChatClient implementace, kde agent používá IChatClient pro odvozování volání, například při použití ChatClientAgent.

Všechny typy middlewaru se implementují prostřednictvím zpětného volání funkce a když je registrováno více instancí middlewaru stejného typu, tvoří řetěz, kde se očekává, že každá instance middlewaru bude volat další v řetězci prostřednictvím poskytnuté nextFunc.

Spuštění agenta a volání typů middlewaru lze v agentu zaregistrovat pomocí tvůrce agentů s existujícím objektem agenta.

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

Důležité

runFunc V ideálním případě by runStreamingFunc mělo být k dispozici obojí. Při poskytování pouze nesílovaného middlewaru ho agent použije pro streamování i volání bez streamování. Streamování se bude spouštět pouze v režimu bez streamování, aby stačit očekávání middlewaru.

Poznámka:

Existuje další přetížení, Use(sharedFunc: ...)které vám umožní poskytnout stejný middleware pro non-streaming a streamování bez blokování streamování. Sdílený middleware ale nebude moct zachytit ani přepsat výstup. Toto přetížení by se mělo použít ve scénářích, kdy potřebujete pouze zkontrolovat nebo upravit vstup před dosažením agenta.

IChatClient middleware lze zaregistrovat na před IChatClient použitím s nástrojem ChatClientAgent, pomocí vzoru tvůrce chatovacího klienta.

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

Výstraha

DefaultAzureCredential je vhodný pro vývoj, ale vyžaduje pečlivé zvážení v produkčním prostředí. V produkčním prostředí zvažte použití konkrétních přihlašovacích údajů (např ManagedIdentityCredential. ) k zabránění problémům s latencí, neúmyslnému testování přihlašovacích údajů a potenciálním bezpečnostním rizikům z náhradních mechanismů.

IChatClient middleware lze také zaregistrovat pomocí factory metody při vytváření agenta prostřednictvím jedné z pomocných metod na klientských knihovnách SDK.

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

Spuštění middlewaru agenta

Tady je příklad spuštění middlewaru agenta, který může kontrolovat a/nebo upravovat vstup a výstup ze spuštění agenta.

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

Spuštění middlewaru streamování agenta

Tady je příklad spuštění middlewaru streamování agenta, který může kontrolovat a/nebo upravovat vstup a výstup ze spuštění streamování agenta.

    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 volání funkcí

Poznámka:

Funkce volání middlewaru se v současné době podporuje pouze v AIAgent případě, že se používá FunctionInvokingChatClientnapříklad ChatClientAgent.

Tady je příklad middlewaru volání funkce, který může zkontrolovat a/nebo upravit volanou funkci a výsledek volání funkce.

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

Smyčku volání funkce je možné ukončit pomocí middlewaru volání funkce nastavením zadané FunctionInvocationContext.Terminate hodnoty true. Tím zabráníte volání funkce ve volání požadavku na službu odvozování obsahující výsledky volání funkce po vyvolání funkce. Pokud během této iterace bylo k dispozici více než jedna funkce pro vyvolání, může také zabránit spuštění všech zbývajících funkcí.

Výstraha

Ukončení smyčky volání funkce může vést k tomu, že historie chatu zůstane v nekonzistentním stavu, například obsahující obsah volání funkce bez obsahu výsledků funkce. To může vést k nepoužitelné historii chatu pro další spuštění.

Middleware IChatClient

Tady je příklad middlewaru chatovacího klienta, který může zkontrolovat nebo upravit vstup a výstup požadavku na službu odvozování, kterou chatovací klient poskytuje.

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

Návod

Kompletní runnable příklady najdete v ukázkách .NET .

Poznámka:

Další informace o middlewaru naleznete v IChatClient tématu Vlastní middleware IChatClient.

Rozhraní Agent Framework je možné přizpůsobit pomocí tří různých typů middlewaru:

1. Middleware agenta: Zachytí spuštění agenta, což vám umožní kontrolovat a upravovat vstupy, výstupy a tok řízení.
2. Middleware funkce: Zachytává volání funkce (nástroj) provedená během provádění agenta, povolení ověřování vstupu, transformace výsledku a řízení provádění.
3. Middleware chatu: Zachytí podkladové žádosti o chat odeslané do modelů AI a poskytuje přístup k nezpracovaným zprávám, možnostem a odpovědím.

Všechny typy podporují implementace založené na funkcích i na třídách. Pokud je zaregistrováno více middlewarů stejného typu, vytvoří řetěz, ve kterém každý volá call_next zpětné volání pro pokračování zpracování. call_next nepřebírají kontext jako argument; middleware ztlumí objekt sdíleného kontextu přímo a pak čeká call_next().

Poznámka:

Pořadí middlewaru se smíšenými obory registrace:

• Middleware na úrovni agenta zabalí middleware na úrovni spuštění.
• Pro middleware agenta a spuštění middleware [A1, A2][R1, R2], pořadí provádění je: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1.
• Middleware funkce/chatu se řídí stejným principem zabalení v době volání pomocí nástroje nebo chatu.

Middleware agenta

Zachytávání middlewaru agenta a úpravy spuštění agenta. Používá následující AgentContext funkce:

• agent: Vyvolání agenta
• messages: Seznam chatovacích zpráv v konverzaci
• session: Aktuální relace agenta, pokud existuje
• options: Možnosti spuštění agenta pro toto vyvolání
• stream: Logická hodnota označující, jestli je odpověď streamovaná
• metadata: Slovník pro ukládání dalších dat mezi middlewarem
• result: Odpověď agenta (lze upravit)
• kwargs: Argumenty klíčového slova starší verze modulu runtime předané metodě spuštění agenta
• client_kwargs: Hodnoty modulu runtime specifické pro klienty podřízeného chatu
• function_invocation_kwargs: Hodnoty modulu runtime, které budou předány do nástrojů

call_next Zpětné volání pokračuje v řetězu middlewaru nebo provádí agenta, pokud se jedná o poslední middleware.

Založené na funkcích

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

Založené na třídách

Middleware agenta založený na třídách používá metodu process , která má stejný podpis a chování jako middleware založený na funkcích.

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

Middleware funkce

Middleware funkcí zachycuje volání funkcí v rámci agentů. Používá následující FunctionInvocationContext funkce:

• function: Vyvolána funkce
• arguments: Ověřené argumenty funkce
• session: Aktuální relace agenta, pokud existuje
• metadata: Slovník pro ukládání dalších dat mezi middlewarem
• result: Návratová hodnota funkce (lze upravit)
• kwargs: Argumenty klíčového slova modulu runtime, které se předávají do vyvolání nástroje.

Zpětné call_next volání pokračuje k dalšímu middlewaru nebo provede skutečnou funkci.

Založené na funkcích

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

Založené na třídách

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

Middleware chatu

Middleware chatu zachycuje žádosti o chat odeslané do modelů AI. Používá následující ChatContext funkce:

• chat_client: Vyvolání chatovacího klienta
• messages: Seznam zpráv odesílaných do služby AI
• options: Možnosti žádosti o chat
• stream: Logická hodnota označující, jestli se jedná o vyvolání streamování
• metadata: Slovník pro ukládání dalších dat mezi middlewarem
• result: Odpověď na chat z umělé inteligence (lze upravit)
• kwargs: Další argumenty klíčových slov předané chatovacímu klientovi
• function_invocation_kwargs: Hodnoty modulu runtime, které budou přeposílány vrstvou chatu pouze s nástroji

call_next Zpětné volání pokračuje do dalšího middlewaru nebo odešle požadavek službě AI.

Poznámka:

Middleware chatu běží uvnitř smyčky volání funkce. To znamená, že se provádí pro každé volání modelu, včetně volání, která odesílají výsledky nástroje zpět do modelu během vícenásobné posloupnosti volání nástrojů.

Založené na funkcích

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

Založené na třídách

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

Dekorátory middlewaru

Dekorátory poskytují explicitní deklaraci typu middlewaru bez nutnosti psaní poznámek. Jsou užitečné, když nepoužíváte poznámky k typu nebo chcete zabránit neshodám typů:

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

Registrace middlewaru

Middleware je možné zaregistrovat na dvou úrovních s různými obory a chováním.

Agent-Level vs. middleware Run-Level

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

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

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

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

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

Klíčové rozdíly:

• Úroveň agenta: Trvalá napříč všemi spuštěními nakonfigurovaná jednou při vytváření agenta
• Úroveň spuštění: Vztahuje se pouze na konkrétní spuštění, umožňuje přizpůsobení jednotlivých požadavků.
• Pořadí spuštění: Middleware agenta (vnější) → Spuštění middlewaru (nejvnitřnější) → spuštění agenta

Ukončení middlewaru

Middleware může ukončit provádění brzy nastavením context.result a zvýšením MiddlewareTermination. To je užitečné pro kontroly zabezpečení, omezování rychlosti nebo chyby ověřování.

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

Co znamená ukončení:

• Nastavení context.result před vyvoláním MiddlewareTermination , pokud chcete vrátit vlastní odpověď
• Zvýšení MiddlewareTermination zastaví zbytek middlewarového řetězce a přeskočí normální cestu provádění.
• Tento model funguje pro middleware agenta, funkce a chatu.

Přepsání výsledku middlewaru

Middleware může přepsat výsledky ve scénářích bez streamování i ve scénářích streamování, což vám umožní změnit nebo zcela nahradit odpovědi agenta.

Typ context.result výsledku závisí na tom, jestli je vyvolání agenta streamované nebo nestreamové:

• Bez streamování: context.result obsahuje AgentResponse úplnou odpověď.
• Streamování: context.result obsahuje asynchronní generátor, který poskytuje AgentResponseUpdate bloky dat.

Můžete použít context.stream k rozlišení těchto scénářů a zpracování přepsání výsledků odpovídajícím způsobem.

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

Tento přístup middlewaru umožňuje implementovat sofistikované transformace odpovědí, filtrování obsahu, vylepšení výsledků a přizpůsobení streamování a přitom udržovat logiku agenta čistou a zaměřenou.

Kompletní příklady middlewaru

Middleware založený na třídě

# 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 založený na funkcích

# 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 založený na dekorátoru

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

Další kroky

Odpovědi na pozadí agenta