共用方式為


代理中介軟體

Agent Framework 中的中介軟體提供了一種強大的方式,能攔截、修改及增強執行各階段的代理互動。 您可以使用中介軟體來實作跨領域關注點,例如日誌記錄、安全驗證、錯誤處理和結果轉換,而無需修改核心代理程式或函數邏輯。

代理框架可透過三種不同類型的中介軟體進行客製化:

  1. 代理程式執行中介件:允許攔截所有代理程式執行,以便可以根據需要檢查和/或修改輸入和輸出。
  2. 函數呼叫中間件:允許攔截代理執行的所有函數調用,以便根據需要檢查和修改輸入和輸出。
  3. IChatClient中介軟體:允許攔截執行執行IChatClientIChatClient者進行推論呼叫時的呼叫,例如當 使用 ChatClientAgent

所有類型的中間件都是透過函數回呼實現的,當註冊多個相同類型的中間件實例時,它們會形成一個鏈,其中每個中間件實例都期望透過提供的 nextFunc.

代理程式執行及函數呼叫中介軟體類型可以在代理程式上登錄,方法是將代理程式建置器與現有代理程式物件搭配使用。

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

這很重要

理想狀況下runFuncrunStreamingFunc兩者都應該提供。 當只提供非串流中介軟體時,代理程式會同時使用它來執行串流與非串流的呼叫。 串流只會在非串流模式下運行,以符合中介軟體的預期。

備註

還有一個額外的過載, Use(sharedFunc: ...)允許你提供相同的中介軟體,適用於非串流和串流,而不阻擋串流。 不過,共用的中介軟體無法攔截或覆蓋輸出。 這種過載應該用於只需檢查或修改輸入,然後才會到達代理的情境。

IChatClient 中介軟體可以在 上 IChatClient 註冊,然後 ChatClientAgent再使用 ,方法是使用 聊天用戶端產生器模式。

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

警告

DefaultAzureCredential 開發方便,但在生產過程中需謹慎考量。 在生產環境中,建議使用特定的憑證(例如 ManagedIdentityCredential),以避免延遲問題、意外的憑證探測,以及備援機制帶來的安全風險。

IChatClient 在透過 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());

代理程式執行中介軟體

以下是代理程式執行中介軟體的範例,可檢查和/或修改代理程式執行的輸入和輸出。

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

代理運行串流中介軟體

這裡有一個代理執行串流中介軟體的範例,可以檢查和/或修改代理串流執行的輸入與輸出。

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

函數呼叫中間件

備註

目前僅支援具備AIAgent的函式呼叫中介軟體,例如FunctionInvokingChatClient

以下是函數呼叫中介軟體的範例,它可以檢查和/或修改正在呼叫的函數,以及函數呼叫的結果。

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

可以透過將提供的 FunctionInvocationContext.Terminate 設定為 true 來終止函數呼叫中介軟體的函數呼叫迴圈。 這可防止函式呼叫迴圈在函式呼叫之後,向包含函式呼叫結果的推論服務發出要求。 如果在此迭代中有多個可調用函式,也可能阻止剩餘函式的執行。

警告

終止函式呼叫迴圈可能會導致聊天紀錄處於不一致狀態,例如包含函式呼叫內容卻沒有函式結果內容。 這可能導致聊天紀錄無法繼續使用。

IChatClient 中介軟體

以下是聊天用戶端中介軟體的範例,可以檢查和/或修改聊天用戶端提供的推論服務請求的輸入和輸出。

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

小提示

完整可執行範例請參閱 .NET 範例

備註

如需中介軟體的詳細資訊 IChatClient ,請參閱 自訂 IChatClient 中介軟體

代理框架可透過三種不同類型的中介軟體進行客製化:

  1. 代理中介軟體:攔截代理執行,讓你能檢查並修改輸入、輸出及控制流程。
  2. 函式中介軟體:攔截代理執行期間所發出的函式(工具)呼叫,實現輸入驗證、結果轉換及執行控制。
  3. 聊天中介軟體:攔截發送給 AI 模型的底層聊天請求,提供原始訊息、選項與回應的存取權限。

所有類型都支援函式基礎與類別基礎實作。 當多個同類型的中介軟體被註冊時,它們會形成一條鏈,每個中介軟體都會呼叫可 next 呼叫物件以繼續處理。

備註

中介軟體順序與混合註冊範圍:

  • 代理層級中介軟體會包裝執行層級的中介軟體。
  • 對於代理中介軟體 [A1, A2] 與執行中介軟體 [R1, R2],執行順序為: A1 -> A2 -> R1 -> R2 -> Agent -> R2 -> R1 -> A2 -> A1
  • 功能/聊天中介軟體在工具/聊天呼叫時遵循相同的包裝原則。

代理中介軟體

代理程式中介軟體會攔截並修改代理程式執行執行。 它使用 which AgentContext 包含:

  • agent:呼叫的代理程式
  • messages:對話中的聊天訊息清單
  • is_streaming:布林值,指出回應是否為串流
  • metadata:用於在中間件之間儲存額外資料的字典
  • result:客服專員的回應(可修改)
  • terminate:旗標以停止進一步處理
  • kwargs:傳遞至代理程式執行方法的其他關鍵字引數

可呼叫者 next 會繼續中介軟體鏈結,或執行代理程式(如果它是最後一個中介軟體)。

基於函數

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

階級基礎

以類別為基礎的代理程式中介軟體使用的方法 process ,其特徵與以函數為基礎的中介軟體具有相同的簽章和行為。

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

函數中間件

函數中介件會攔截代理程式內的函數呼叫。 它使用 which FunctionInvocationContext 包含:

  • function:正在呼叫的函數
  • arguments:函數的已驗證引數
  • metadata:用於在中間件之間儲存額外資料的字典
  • result:函數的傳回值(可修改)
  • terminate:旗標以停止進一步處理
  • kwargs:傳遞至叫用此函式之聊天方法的其他關鍵字引數

可呼叫繼續 next 到下一個中間件或執行實際函數。

基於函數

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

階級基礎

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

聊天中間件

聊天中間件攔截發送給 AI 模型的聊天請求。 它使用 which ChatContext 包含:

  • chat_client:正在叫用的聊天用戶端
  • messages:傳送至 AI 服務的訊息清單
  • options:聊天請求的選項
  • is_streaming:布林值,指出這是否是串流呼叫
  • metadata:用於在中間件之間儲存額外資料的字典
  • result:來自 AI 的聊天回應(可修改)
  • terminate:旗標以停止進一步處理
  • kwargs:傳遞至聊天用戶端的其他關鍵字引數

可呼叫項目會繼續 next 到下一個中介軟體,或將要求傳送至 AI 服務。

基於函數

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

階級基礎

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

中介軟體裝飾器

裝飾器提供明確的中介軟體類型宣告,而不需要類型註解。 當你不使用類型註解或想避免類型不匹配時,它們很有幫助:

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

中間件註冊

中間件可以在兩個層級註冊,具有不同的範圍和行為。

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

主要區別:

  • 代理程式層級:在所有執行中持續存在,在建立代理程式時設定一次
  • 執行層級:僅套用至特定執行,允許每個要求自訂
  • 執行順序:代理中間件(最外層)→運行中間件(最內層)→代理執行

中間件終止

中間件可以使用 提前終止執行 context.terminate。 這對於安全性檢查、速率限制或驗證失敗非常有用。

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)

終止的意義:

  • 設定 context.terminate = True 處理應停止的訊號
  • 您可以在終止之前提供自訂結果,以提供使用者意見反應
  • 當中間件終止時,會完全略過代理程式執行

中介軟體結果覆寫

中介軟體可以覆寫非串流和串流案例中的結果,可讓您修改或完全取代代理程式回應。

結果 context.result 類型取決於代理程式呼叫是串流還是非串流:

  • 非串流:context.result包含具有AgentResponse完整回應的
  • 串流context.result 包含產生區塊的 AgentResponseUpdate 非同步產生器

您可以使用來 context.is_streaming 區分這些案例,並適當地處理結果覆寫。

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

這種中間件方法允許您實施複雜的響應轉換、內容過濾、結果增強和流自定義,同時保持代理邏輯乾淨且集中。

完整中介軟體範例

基於類別的中介軟體

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

功能式中介軟體

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

基於裝飾器的中介軟體

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

後續步驟