Agent Framework 中的中介軟體提供了一種強大的方式,能攔截、修改及增強執行各階段的代理互動。 您可以使用中介軟體來實作跨領域關注點,例如日誌記錄、安全驗證、錯誤處理和結果轉換,而無需修改核心代理程式或函數邏輯。
代理框架可透過三種不同類型的中介軟體進行客製化:
- 代理程式執行中介件:允許攔截所有代理程式執行,以便可以根據需要檢查和/或修改輸入和輸出。
- 函數呼叫中間件:允許攔截代理執行的所有函數調用,以便根據需要檢查和修改輸入和輸出。
-
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 中介軟體。
代理框架可透過三種不同類型的中介軟體進行客製化:
- 代理中介軟體:攔截代理執行,讓你能檢查並修改輸入、輸出及控制流程。
- 函式中介軟體:攔截代理執行期間所發出的函式(工具)呼叫,實現輸入驗證、結果轉換及執行控制。
- 聊天中介軟體:攔截發送給 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())