Middleware für Agenten

Middleware in Agent Framework bietet eine leistungsstarke Möglichkeit, Agentinteraktionen in verschiedenen Phasen der Ausführung abzufangen, zu ändern und zu verbessern. Sie können Middleware verwenden, um übergreifende Bedenken wie Protokollierung, Sicherheitsüberprüfung, Fehlerbehandlung und Ergebnistransformation zu implementieren, ohne Ihre Kern-Agent- oder Funktionslogik zu ändern.

Agent Framework kann mithilfe von drei verschiedenen Middlewaretypen angepasst werden:

Agent Run Middleware: Ermöglicht das Abfangen aller Agentausführungen, sodass Eingabe und Ausgabe bei Bedarf überprüft und/oder geändert werden können.
Middleware für Funktionsaufrufe: Ermöglicht das Abfangen aller Funktionsaufrufe, die vom Agent ausgeführt werden, sodass Eingabe und Ausgabe bei Bedarf überprüft und geändert werden können.
IChatClient Middleware: Ermöglicht das Abfangen von Aufrufen an eine IChatClient Implementierung, bei der ein Agent IChatClient für Ableitungsaufrufe verwendet wird, z. B. bei verwendung ChatClientAgent.

Alle Middlewaretypen werden über einen Funktionsrückruf implementiert, und wenn mehrere Middlewareinstanzen desselben Typs registriert werden, bilden sie eine Kette, in der jede Middleware-Instanz die nächste in der Kette über eine bereitgestellte nextFuncInstanz aufrufen soll.

Agent run and function calling middleware types can be registered on an agent, by using the agent builder with an existing agent object.

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

Von Bedeutung

Idealerweise sollten beide runFunc bereitgestellt runStreamingFunc werden. Wenn Sie nur die Nicht-Streaming-Middleware bereitstellen, verwendet der Agent sie sowohl für Streaming- als auch für Nicht-Streaming-Aufrufe. Streaming wird nur im Nicht-Streaming-Modus ausgeführt, um die Erwartungen an Middleware zu erfüllen.

Hinweis

Es gibt eine zusätzliche Überladung, Use(sharedFunc: ...)mit der Sie die gleiche Middleware für Nicht-Streaming und Streaming bereitstellen können, ohne das Streaming zu blockieren. Die freigegebene Middleware kann die Ausgabe jedoch nicht abfangen oder außer Kraft setzen. Diese Überladung sollte für Szenarien verwendet werden, in denen Sie die Eingabe nur überprüfen oder ändern müssen, bevor sie den Agent erreicht.

IChatClientMiddleware kann vor der Verwendung mit einem IChatClientClient-Generator-Muster für Chats registriert ChatClientAgent werden.

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

Warnung

DefaultAzureCredential ist praktisch für die Entwicklung, erfordert aber sorgfältige Überlegungen in der Produktion. Berücksichtigen Sie in der Produktion die Verwendung bestimmter Anmeldeinformationen (z. B. ManagedIdentityCredential), um Latenzprobleme, unbeabsichtigte Anmeldeinformationen probingen und potenzielle Sicherheitsrisiken aus Fallbackmechanismen zu vermeiden.

IChatClient Middleware kann auch mithilfe einer Factorymethode registriert werden, wenn ein Agent über eine der Hilfsmethoden auf SDK-Clients erstellt wird.

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

Agent Run Middleware

Nachfolgend finden Sie ein Beispiel für die Middleware des Agents, die die Eingabe und/oder Ausgabe des Agents überprüfen und/oder ändern kann.

async Task<AgentResponse> CustomAgentRunMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    Console.WriteLine(messages.Count());
    var response = await innerAgent.RunAsync(messages, session, options, cancellationToken).ConfigureAwait(false);
    Console.WriteLine(response.Messages.Count);
    return response;
}

Agent Run Streaming Middleware

Hier sehen Sie ein Beispiel für die Agentausführung von Streaming-Middleware, die die Eingabe und/oder Ausgabe der Agent-Streamingausführung prüfen und/oder ändern kann.

    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 für Funktionsaufrufe

Hinweis

Die Middleware für Funktionsaufrufe wird derzeit nur mit einem AIAgent unterstützt, das FunctionInvokingChatClient verwendet, wie beispielsweise ChatClientAgent.

Nachfolgend finden Sie ein Beispiel für die Middleware zum Aufrufen von Funktionen, die die aufgerufene Funktion prüfen und/oder ändern können, und das Ergebnis des Funktionsaufrufs.

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

Es ist möglich, die Funktionsaufrufschleife mit Funktionsaufrufs-Middleware zu beenden, indem sie die angegebene FunctionInvocationContext.Terminate Aufschrift auf "true" festlegen. Dadurch wird verhindert, dass die Funktionsaufrufschleife nach dem Aufruf der Funktion eine Anforderung an den Rückschlussdienst ausgibt, der die Ergebnisse des Funktionsaufrufs enthält. Wenn während dieser Iteration mehrere Funktionen zum Aufrufen verfügbar waren, kann es auch verhindern, dass verbleibende Funktionen ausgeführt werden.

Warnung

Das Beenden der Funktionsaufrufschleife kann dazu führen, dass der Chatverlauf in einem inkonsistenten Zustand verbleibt, z. B. mit Funktionsaufrufinhalten ohne Funktionsergebnisinhalt. Dies kann dazu führen, dass der Chatverlauf für weitere Ausführungen nicht mehr verwendet werden kann.

IChatClient-Middleware

Hier ist ein Beispiel für Chatclient-Middleware, die die Eingabe und/oder Ausgabe für die Anforderung an den vom Chatclient bereitgestellten Ableitungsdienst überprüfen und/oder ändern kann.

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

Tipp

Vollständige Runnable-Beispiele finden Sie in den .NET-Beispielen .

Hinweis

Weitere Informationen zu IChatClient Middleware finden Sie unter Custom IChatClient middleware.

Agent Framework kann mithilfe von drei verschiedenen Middlewaretypen angepasst werden:

Agent-Middleware: Abfangen der Ausführung des Agents, sodass Sie Eingaben, Ausgaben und Steuerungsfluss prüfen und ändern können.
Funktions-Middleware: Abfangen von Funktionsaufrufen (Tool) während der Agentausführung, Aktivieren der Eingabeüberprüfung, Ergebnistransformation und Ausführungssteuerung.
Chat-Middleware: Fängt die zugrunde liegenden Chatanfragen ab, die an KI-Modelle gesendet werden, und bietet Zugriff auf die unformatierten Nachrichten, Optionen und Antworten.

Alle Typen unterstützen sowohl funktionsbasierte als auch klassenbasierte Implementierungen. Wenn mehrere Middleware desselben Typs registriert sind, bilden sie eine Kette, in der die einzelnen Aufrufe aufgerufen werden, um die next Verarbeitung fortzusetzen.

Middleware für Agenten

Agent Middleware fängt ab und ändert die Ausführung des Agents. Es verwendet folgendes AgentContext :

agent: Der aufgerufene Agent
messages: Liste der Chatnachrichten in der Unterhaltung
is_streaming: Boolescher Wert, der angibt, ob die Antwort streamingt
metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
result: Die Antwort des Agents (kann geändert werden)
terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
kwargs: Zusätzliche Schlüsselwortargumente, die an die Agent-Run-Methode übergeben werden

Der next Aufrufbare setzt die Middlewarekette fort oder führt den Agent aus, wenn es sich um die letzte Middleware handelt.

Funktionsbasiert

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

Klassenbasiert

Klassenbasierte Agent-Middleware verwendet eine process Methode, die dieselbe Signatur und dasselbe Verhalten wie funktionsbasierte Middleware aufweist.

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

Funktions-Middleware

Funktions-Middleware fängt Funktionsaufrufe innerhalb von Agents ab. Es verwendet folgendes FunctionInvocationContext :

function: Die aufgerufene Funktion
arguments: Die überprüften Argumente für die Funktion
metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
result: Der Rückgabewert der Funktion (kann geändert werden)
terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
kwargs: Zusätzliche Schlüsselwortargumente, die an die Chatmethode übergeben werden, die diese Funktion aufgerufen hat

Der next Aufrufbare setzt die nächste Middleware fort oder führt die eigentliche Funktion aus.

Funktionsbasiert

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

Klassenbasiert

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

Chat-Middleware

Chat-Middleware fängt Chatanfragen ab, die an KI-Modelle gesendet werden. Es verwendet folgendes ChatContext :

chat_client: Der aufgerufene Chatclient
messages: Liste der Nachrichten, die an den KI-Dienst gesendet werden
options: Die Optionen für die Chatanfrage
is_streaming: Boolescher Wert, der angibt, ob es sich um einen Streamingaufruf handelt
metadata: Wörterbuch zum Speichern zusätzlicher Daten zwischen Middleware
result: Die Chatantwort von der KI (kann geändert werden)
terminate: Kennzeichnung, um die weitere Verarbeitung zu beenden
kwargs: Zusätzliche Schlüsselwortargumente, die an den Chatclient übergeben werden

Der next Aufrufbare wird weiterhin an die nächste Middleware gesendet oder die Anforderung an den KI-Dienst gesendet.

Funktionsbasiert

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

Klassenbasiert

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

Middleware-Dekoratoren

Dekorateure stellen explizite Middlewaretypdeklaration bereit, ohne Dass Typanmerkungen erforderlich sind. Sie sind hilfreich, wenn Sie keine Typanmerkungen verwenden oder Typenkonflikte verhindern möchten:

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

Middleware-Registrierung

Middleware kann auf zwei Ebenen mit unterschiedlichen Bereichen und Verhaltensweisen registriert werden.

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

Wichtige Unterschiede:

Agent-Ebene: Persistent für alle Ausgeführten, einmal beim Erstellen des Agents konfiguriert
Ausführungsebene: Nur auf bestimmte Ausführungen angewendet, ermöglicht Anpassungen pro Anforderung
Ausführungsreihenfolge: Agent Middleware (äußerst) → Middleware ausführen (innerstes) → Agentausführung

Middleware-Beendigung

Middleware kann die Ausführung frühzeitig beenden.context.terminate Dies ist nützlich für Sicherheitsüberprüfungen, Ratenbeschränkungen oder Überprüfungsfehler.

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)

Was eine Kündigung bedeutet:

Festlegen von context.terminate = True Signalen, dass die Verarbeitung beendet werden soll
Sie können ein benutzerdefiniertes Ergebnis vor dem Beenden bereitstellen, um Benutzern Feedback zu geben.
Die Agentausführung wird vollständig übersprungen, wenn Middleware beendet wird.

Außerkraftsetzung von Middleware-Ergebnissen

Middleware kann Ergebnisse sowohl in Nicht-Streaming- als auch Streamingszenarien außer Kraft setzen, sodass Sie Agent-Antworten ändern oder vollständig ersetzen können.

Der Ergebnistyp context.result hängt davon ab, ob der Agentaufruf Streaming oder Nicht-Streaming ist:

Nicht-Streaming: context.result enthält eine AgentResponse mit der vollständigen Antwort
Streaming: context.result enthält einen asynchronen AgentResponseUpdate Generator, der Blöcke liefert

Sie können verwenden context.is_streaming , um zwischen diesen Szenarien zu unterscheiden und Ergebnisüberschreibungen entsprechend zu behandeln.

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

Mit diesem Middleware-Ansatz können Sie anspruchsvolle Antworttransformationen, Inhaltsfilterung, Ergebnisverbesserungen und Streaminganpassungen implementieren und gleichzeitig die Agentlogik sauber und fokussiert halten.

Vollständige Middleware-Beispiele

Klassenbasierte Middleware

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Funktionsbasierte Middleware

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Dekorateur-basierte Middleware

# Copyright (c) Microsoft. All rights reserved.

import asyncio
import time
from collections.abc import Awaitable, Callable
from random import randint
from typing import Annotated

from agent_framework import (
    AgentContext,
    AgentMiddleware,
    AgentResponse,
    FunctionInvocationContext,
    FunctionMiddleware,
    Message,
    tool,
)
from agent_framework.azure import AzureAIAgentClient
from azure.identity.aio import AzureCliCredential
from pydantic import Field

"""
Class-based MiddlewareTypes Example

This sample demonstrates how to implement middleware using class-based approach by inheriting
from AgentMiddleware and FunctionMiddleware base classes. The example includes:

- SecurityAgentMiddleware: Checks for security violations in user queries and blocks requests
  containing sensitive information like passwords or secrets
- LoggingFunctionMiddleware: Logs function execution details including timing and parameters

This approach is useful when you need stateful middleware or complex logic that benefits
from object-oriented design patterns.
"""


# NOTE: approval_mode="never_require" is for sample brevity. Use "always_require" in production; see samples/02-agents/tools/function_tool_with_approval.py and samples/02-agents/tools/function_tool_with_approval_and_sessions.py.
@tool(approval_mode="never_require")
def get_weather(
    location: Annotated[str, Field(description="The location to get the weather for.")],
) -> str:
    """Get the weather for a given location."""
    conditions = ["sunny", "cloudy", "rainy", "stormy"]
    return f"The weather in {location} is {conditions[randint(0, 3)]} with a high of {randint(10, 30)}°C."


class SecurityAgentMiddleware(AgentMiddleware):
    """Agent middleware that checks for security violations."""

    async def process(
        self,
        context: AgentContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        # Check for potential security violations in the query
        # Look at the last user message
        last_message = context.messages[-1] if context.messages else None
        if last_message and last_message.text:
            query = last_message.text
            if "password" in query.lower() or "secret" in query.lower():
                print("[SecurityAgentMiddleware] Security Warning: Detected sensitive information, blocking request.")
                # Override the result with warning message
                context.result = AgentResponse(
                    messages=[Message("assistant", ["Detected sensitive information, the request is blocked."])]
                )
                # Simply don't call call_next() to prevent execution
                return

        print("[SecurityAgentMiddleware] Security check passed.")
        await call_next()


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

    async def process(
        self,
        context: FunctionInvocationContext,
        call_next: Callable[[], Awaitable[None]],
    ) -> None:
        function_name = context.function.name
        print(f"[LoggingFunctionMiddleware] About to call function: {function_name}.")

        start_time = time.time()

        await call_next()

        end_time = time.time()
        duration = end_time - start_time

        print(f"[LoggingFunctionMiddleware] Function {function_name} completed in {duration:.5f}s.")


async def main() -> None:
    """Example demonstrating class-based middleware."""
    print("=== Class-based MiddlewareTypes Example ===")

    # For authentication, run `az login` command in terminal or replace AzureCliCredential with preferred
    # authentication option.
    async with (
        AzureCliCredential() as credential,
        AzureAIAgentClient(credential=credential).as_agent(
            name="WeatherAgent",
            instructions="You are a helpful weather assistant.",
            tools=get_weather,
            middleware=[SecurityAgentMiddleware(), LoggingFunctionMiddleware()],
        ) as agent,
    ):
        # Test with normal query
        print("\n--- Normal Query ---")
        query = "What's the weather like in Seattle?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")

        # Test with security-related query
        print("--- Security Test ---")
        query = "What's the password for the weather service?"
        print(f"User: {query}")
        result = await agent.run(query)
        print(f"Agent: {result.text}\n")


if __name__ == "__main__":
    asyncio.run(main())

Nächste Schritte

Agent-Hintergrundantworten

Feedback

War diese Seite hilfreich?

Last updated on 2026-02-13

Freigeben über

Middleware für Agenten

Agent Run Middleware

Agent Run Streaming Middleware

Middleware für Funktionsaufrufe

IChatClient-Middleware

Middleware für Agenten

Funktionsbasiert

Klassenbasiert

Funktions-Middleware

Funktionsbasiert

Klassenbasiert

Chat-Middleware

Funktionsbasiert

Klassenbasiert

Middleware-Dekoratoren

Middleware-Registrierung

Agent-Level vs Run-Level Middleware

Middleware-Beendigung

Außerkraftsetzung von Middleware-Ergebnissen

Vollständige Middleware-Beispiele

Klassenbasierte Middleware

Funktionsbasierte Middleware

Dekorateur-basierte Middleware

Nächste Schritte

Feedback

Zusätzliche Ressourcen