에이전트 프레임워크의 미들웨어는 다양한 실행 단계에서 에이전트 상호 작용을 가로채고 수정하고 개선하는 강력한 방법을 제공합니다. 미들웨어를 사용하여 핵심 에이전트 또는 함수 논리를 수정하지 않고 로깅, 보안 유효성 검사, 오류 처리 및 결과 변환과 같은 교차 절삭 문제를 구현할 수 있습니다.
에이전트 프레임워크는 세 가지 유형의 미들웨어를 사용하여 사용자 지정할 수 있습니다.
- 에이전트 실행 미들웨어: 필요에 따라 입력 및 출력을 검사 및/또는 수정할 수 있도록 모든 에이전트 실행을 차단할 수 있습니다.
- 함수 호출 미들웨어: 필요에 따라 입력 및 출력을 검사하고 수정할 수 있도록 에이전트에서 실행하는 모든 함수 호출의 가로채기를 허용합니다.
-
IChatClient 미들웨어: 예를 들어 사용 시 에이전트가 유추 호출
IChatClient에 사용하는IChatClient구현에 대한 호출을 차단할 수 있습니다ChatClientAgent.
모든 유형의 미들웨어는 함수 콜백을 통해 구현되며, 동일한 형식의 여러 미들웨어 인스턴스가 등록되면 체인을 형성합니다. 여기서 각 미들웨어 인스턴스는 제공된 nextFunc것을 통해 체인의 다음 인스턴스를 호출해야 합니다.
에이전트 실행 및 함수 호출 미들웨어 형식은 기존 에이전트 개체와 함께 에이전트 작성기를 사용하여 에이전트에 등록할 수 있습니다.
var middlewareEnabledAgent = originalAgent
.AsBuilder()
.Use(runFunc: CustomAgentRunMiddleware, runStreamingFunc: CustomAgentRunStreamingMiddleware)
.Use(CustomFunctionCallingMiddleware)
.Build();
중요합니다
이상적으로 둘 다 runFunc 제공 runStreamingFunc 되어야 합니다. 비 스트리밍 미들웨어만 제공하는 경우 에이전트는 스트리밍 및 비 스트리밍 호출 모두에 사용합니다. 스트리밍은 스트리밍이 아닌 모드에서만 실행되어 미들웨어 기대에 충분합니다.
비고
스트리밍을 차단하지 않고 비 스트리밍 및 스트리밍에 동일한 미들웨어를 제공할 수 있는 추가 오버로드 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만이 함수 호출 미들웨어를 현재 지원합니다, 예를 들어 ChatClientAgent.
다음은 호출되는 함수와 함수 호출의 결과를 검사 및/또는 수정할 수 있는 함수 호출 미들웨어의 예입니다.
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 하여 처리를 계속하는 체인을 형성합니다.
에이전트 미들웨어
에이전트 미들웨어는 에이전트 실행 실행을 가로채고 수정합니다. 다음을 포함하는 항목을 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")
함수 미들웨어
함수 미들웨어는 에이전트 내의 함수 호출을 차단합니다. 다음을 포함하는 항목을 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 모델로 전송된 채팅 요청을 차단합니다. 다음을 포함하는 항목을 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())