Agen Middleware

Middleware dalam Agent Framework menyediakan cara yang ampuh untuk mencegat, memodifikasi, dan meningkatkan interaksi agen pada berbagai tahap eksekusi. Anda dapat menggunakan middleware untuk menerapkan masalah lintas pemotongan seperti pengelogan, validasi keamanan, penanganan kesalahan, dan transformasi hasil tanpa memodifikasi agen inti atau logika fungsi Anda.

Agent Framework dapat disesuaikan menggunakan tiga jenis middleware yang berbeda:

1. Middleware Eksekusi Agen: Memungkinkan intersepsi semua eksekusi agen, sehingga input dan output dapat diperiksa dan/atau dimodifikasi sesuai kebutuhan.
2. Middleware panggilan fungsi: Memungkinkan intersepsi semua panggilan fungsi yang dijalankan oleh agen, sehingga input dan output dapat diperiksa dan dimodifikasi sesuai kebutuhan.
3. IChatClient middleware: Memungkinkan intersepsi panggilan ke IChatClient implementasi, di mana agen menggunakan IChatClient untuk panggilan inferensi, misalnya, saat menggunakan ChatClientAgent.

Semua jenis middleware diimplementasikan melalui panggilan balik fungsi, dan ketika beberapa instans middleware dari jenis yang sama terdaftar, mereka membentuk rantai, di mana setiap instans middleware diharapkan untuk memanggil berikutnya dalam rantai, melalui yang disediakan nextFunc.

Agen menjalankan dan memanggil jenis middleware fungsi dapat didaftarkan pada agen, dengan menggunakan pembuat agen dengan objek agen yang ada.

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

Penting

Idealnya baik runFunc dan runStreamingFunc harus disediakan. Saat hanya menyediakan middleware non-streaming, agen akan menggunakannya untuk pemanggilan streaming dan non-streaming. Streaming hanya akan berjalan dalam mode non-streaming untuk mencukupkan harapan middleware.

Nota

Ada kelebihan beban tambahan, Use(sharedFunc: ...), yang memungkinkan Anda menyediakan middleware yang sama untuk non-streaming dan streaming tanpa memblokir streaming. Namun, middleware bersama tidak akan dapat mencegat atau mengambil alih output. Kelebihan beban ini harus digunakan untuk skenario di mana Anda hanya perlu memeriksa atau memodifikasi input sebelum mencapai agen.

IChatClient middleware dapat didaftarkan pada IChatClient sebelum digunakan dengan ChatClientAgent, dengan menggunakan pola pembangun klien obrolan.

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

Peringatan

DefaultAzureCredential nyaman untuk pengembangan tetapi membutuhkan pertimbangan yang cermat dalam produksi. Dalam produksi, pertimbangkan untuk menggunakan kredensial tertentu (misalnya, ManagedIdentityCredential) untuk menghindari masalah latensi, pemeriksaan kredensial yang tidak diinginkan, dan potensi risiko keamanan dari mekanisme fallback.

IChatClient middleware juga dapat didaftarkan menggunakan metode pabrik saat membuat agen melalui salah satu metode pembantu pada klien 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());

Middleware Eksekusi Agen

Berikut adalah contoh middleware eksekusi agen, yang dapat memeriksa dan/atau memodifikasi input dan output dari agen yang dijalankan.

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

Middleware Streaming Eksekusi Agen

Berikut adalah contoh middleware streaming eksekusi agen, yang dapat memeriksa dan/atau memodifikasi input dan output dari eksekusi streaming agen.

    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 panggilan fungsi

Nota

Pemanggilan fungsi middleware saat ini hanya didukung dengan AIAgent yang menggunakan FunctionInvokingChatClient, misalnya, ChatClientAgent.

Berikut adalah contoh fungsi memanggil middleware, yang dapat memeriksa dan/atau memodifikasi fungsi yang dipanggil, dan hasil dari panggilan fungsi.

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

Dimungkinkan untuk mengakhiri perulangan panggilan fungsi dengan middleware panggilan fungsi dengan mengatur yang disediakan FunctionInvocationContext.Terminate ke true. Ini akan mencegah perulangan panggilan fungsi mengeluarkan permintaan ke layanan inferensi yang berisi hasil panggilan fungsi setelah pemanggilan fungsi. Jika ada lebih dari satu fungsi yang tersedia untuk pemanggilan selama iterasi ini, mungkin juga mencegah fungsi yang tersisa dijalankan.

Peringatan

Mengakhiri perulangan panggilan fungsi dapat mengakibatkan riwayat obrolan Anda dibiarkan dalam status tidak konsisten, misalnya, berisi konten panggilan fungsi tanpa konten hasil fungsi. Ini dapat mengakibatkan riwayat obrolan tidak dapat digunakan untuk eksekusi lebih lanjut.

Middleware IChatClient

Berikut adalah contoh middleware klien obrolan, yang dapat memeriksa dan/atau memodifikasi input dan output untuk permintaan ke layanan inferensi yang disediakan klien obrolan.

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

Petunjuk / Saran

Lihat sampel .NET untuk contoh lengkap yang dapat dijalankan.

Nota

Untuk informasi selengkapnya tentang IChatClient middleware, lihat Middleware IChatClient kustom.

Agent Framework dapat disesuaikan menggunakan tiga jenis middleware yang berbeda:

1. Middleware agen: Mencegat agen menjalankan eksekusi, memungkinkan Anda memeriksa dan memodifikasi input, output, dan alur kontrol.
2. Middleware fungsi: Mencegat panggilan fungsi (alat) yang dilakukan selama eksekusi agen, mengaktifkan validasi input, transformasi hasil, dan kontrol eksekusi.
3. Middleware obrolan: Mencegat permintaan obrolan yang mendasar yang dikirim ke model AI, menyediakan akses ke pesan mentah, opsi, dan respons.

Semua jenis mendukung implementasi berbasis fungsi dan berbasis kelas. Ketika beberapa middleware dari jenis yang sama terdaftar, mereka membentuk rantai di mana setiap panggilan next dapat dipanggil untuk melanjutkan pemrosesan.

Agen Middleware

Middleware agen mencegat dan memodifikasi agen menjalankan eksekusi. Ini menggunakan yang AgentContext berisi:

• agent: Agen yang sedang dipanggil
• messages: Daftar pesan obrolan dalam percakapan
• is_streaming: Boolean menunjukkan apakah respons sedang streaming
• metadata: Kamus untuk menyimpan data tambahan antar middleware
• result: Respons agen (dapat dimodifikasi)
• terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
• kwargs: Argumen kata kunci tambahan yang diteruskan ke metode eksekusi agen

Yang next dapat dipanggil melanjutkan rantai middleware atau menjalankan agen jika itu middleware terakhir.

Berbasis fungsi

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

Berbasis kelas

Middleware agen berbasis kelas menggunakan process metode yang memiliki tanda tangan dan perilaku yang sama dengan middleware berbasis fungsi.

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

Middleware Fungsi

Middleware fungsi mencegat panggilan fungsi dalam agen. Ini menggunakan yang FunctionInvocationContext berisi:

• function: Fungsi yang sedang dipanggil
• arguments: Argumen yang divalidasi untuk fungsi
• metadata: Kamus untuk menyimpan data tambahan antar middleware
• result: Nilai pengembalian fungsi (dapat dimodifikasi)
• terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
• kwargs: Argumen kata kunci tambahan diteruskan ke metode obrolan yang memanggil fungsi ini

Yang next dapat dipanggil berlanjut ke middleware berikutnya atau menjalankan fungsi aktual.

Berbasis fungsi

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

Berbasis kelas

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

Middleware Obrolan

Middleware obrolan mencegat permintaan obrolan yang dikirim ke model AI. Ini menggunakan yang ChatContext berisi:

• chat_client: Klien obrolan yang dipanggil
• messages: Daftar pesan yang dikirim ke layanan AI
• options: Opsi untuk permintaan obrolan
• is_streaming: Boolean menunjukkan apakah ini adalah pemanggilan streaming
• metadata: Kamus untuk menyimpan data tambahan antar middleware
• result: Respons obrolan dari AI (dapat dimodifikasi)
• terminate: Bendera untuk menghentikan pemrosesan lebih lanjut
• kwargs: Argumen kata kunci tambahan diteruskan ke klien obrolan

Yang next dapat dipanggil berlanjut ke middleware berikutnya atau mengirim permintaan ke layanan AI.

Berbasis fungsi

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

Berbasis kelas

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

Dekorator Middleware

Dekorator menyediakan deklarasi jenis middleware eksplisit tanpa memerlukan anotasi jenis. Ini berguna ketika Anda tidak menggunakan anotasi jenis atau ingin mencegah ketidakcocokan jenis:

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

Pendaftaran Middleware

Middleware dapat didaftarkan pada dua tingkat dengan cakupan dan perilaku yang berbeda.

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

Perbedaan Utama:

• Tingkat agen: Persisten di semua eksekusi, dikonfigurasi sekali saat membuat agen
• Tingkat eksekusi: Hanya diterapkan ke eksekusi tertentu, memungkinkan penyesuaian per permintaan
• Urutan Eksekusi: Middleware agen (terluar) → Jalankan middleware (terdahulu) → eksekusi Agen

Penghentian Middleware

Middleware dapat mengakhiri eksekusi lebih awal menggunakan context.terminate. Ini berguna untuk pemeriksaan keamanan, pembatasan tarif, atau kegagalan validasi.

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)

Apa artinya penghentian:

• Mengatur context.terminate = True sinyal bahwa pemrosesan harus berhenti
• Anda dapat memberikan hasil kustom sebelum mengakhiri untuk memberikan umpan balik kepada pengguna
• Eksekusi agen benar-benar dilewati ketika middleware berakhir

Penimpaan Hasil Middleware

Middleware dapat mengambil alih hasil dalam skenario non-streaming dan streaming, memungkinkan Anda untuk memodifikasi atau mengganti respons agen sepenuhnya.

Jenis hasil tergantung context.result pada apakah pemanggilan agen streaming atau non-streaming:

• Non-streaming: context.result berisi AgentResponse dengan respons lengkap
• Streaming: context.result berisi generator asinkron yang menghasilkan AgentResponseUpdate gugus

Anda dapat menggunakan context.is_streaming untuk membedakan antara skenario ini dan menangani penimpaan hasil dengan tepat.

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

Pendekatan middleware ini memungkinkan Anda menerapkan transformasi respons canggih, pemfilteran konten, peningkatan hasil, dan kustomisasi streaming sambil menjaga logika agen Anda tetap bersih dan fokus.

Contoh middleware lengkap

Middleware berbasis kelas

# 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 berbasis fungsi

# 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 berbasis dekorator

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

Langkah selanjutnya

Respons Latar Belakang Agen