From 9bd4e26eace25d6ef42a4f1183ab589457f0d294 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Thu, 28 May 2026 16:01:30 +0530 Subject: [PATCH 1/2] feat(sample): add middleware pattern for message interception between agents (#3556) --- .../group_chat_agent_manager.py | 162 +++++++++++++++--- 1 file changed, 135 insertions(+), 27 deletions(-) diff --git a/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py b/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py index 079f66ba5c..803c6eae9d 100644 --- a/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py +++ b/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py @@ -2,11 +2,12 @@ import asyncio import os -from typing import cast +from typing import AsyncIterator, Callable, Optional, cast from agent_framework import ( Agent, AgentResponseUpdate, + Event, Message, ) from agent_framework.foundry import FoundryChatClient @@ -17,18 +18,128 @@ # Load environment variables from .env file load_dotenv() +MessageInterceptor = Callable[[Message], Optional[Message]] + + +def compose_middleware(*middlewares: MessageInterceptor) -> MessageInterceptor: + """Chain multiple interceptors. Execution stops early if any returns None.""" + def chained_interceptor(msg: Message) -> Optional[Message]: + current = msg + for mw in middlewares: + if current is None: + break + current = mw(current) + return current + return chained_interceptor + + +# 1. CONTENT FILTERING: Drop or redact messages containing sensitive terms +def create_content_filter(forbidden_terms: list[str], redact_instead: bool = True) -> MessageInterceptor: + def interceptor(msg: Message) -> Optional[Message]: + if msg.role != "assistant" or not msg.text: + return msg + + text_lower = msg.text.lower() + if any(term.lower() in text_lower for term in forbidden_terms): + if redact_instead: + safe_text = "[REDACTED BY MIDDLEWARE]" + try: + return msg.replace(text=safe_text) if hasattr(msg, "replace") else Message( + role=msg.role, author_name=msg.author_name, text=safe_text + ) + except Exception: + return None + print(f"\n[MIDDLEWARE] Content Filter: Dropped message from '{msg.author_name}'") + return None + return msg + return interceptor + + +# 2. MESSAGE TRANSFORMATION: Add metadata prefixes for audit/tracing +def create_message_tagger(prefix: str) -> MessageInterceptor: + def interceptor(msg: Message) -> Optional[Message]: + if msg.role == "user" or not msg.text: + return msg + + new_text = f"[{prefix}] {msg.text}" + try: + return msg.replace(text=new_text) if hasattr(msg, "replace") else Message( + role=msg.role, author_name=msg.author_name, text=new_text + ) + except Exception as e: + print(f"\n[MIDDLEWARE] Transformation failed: {e}") + return msg + return interceptor + + +# 3. ACCESS CONTROL: Restrict messages from specific agents +def create_sender_blocklist(blocked_agents: set[str]) -> MessageInterceptor: + def interceptor(msg: Message) -> Optional[Message]: + if msg.author_name in blocked_agents: + print(f"\n[MIDDLEWARE] Access Control: Blocked message from '{msg.author_name}'") + return None + return msg + return interceptor + +# WORKFLOW WRAPPER +async def run_with_message_middleware( + workflow, + task: str, + middleware: MessageInterceptor, + *, + stream: bool = True, +) -> AsyncIterator[Event]: + """ + Wraps workflow execution to apply message interception. + This is a user-space pattern until native builder hooks are added. + """ + async for event in workflow.run(task, stream=stream): + if event.type == "intermediate" and isinstance(event.data, AgentResponseUpdate): + original_text = event.data.text + if original_text: + temp_msg = Message( + role=event.data.role or "assistant", + author_name=event.data.author_name, + text=original_text + ) + intercepted = middleware(temp_msg) + if intercepted is None: + continue + event.data.text = intercepted.text + + elif event.type == "output" and isinstance(event.data, list): + filtered_messages = [] + for msg in event.data: + if isinstance(msg, Message): + processed = middleware(msg) + if processed is not None: + filtered_messages.append(processed) + event.data = filtered_messages + + yield event + + """ -Sample: Group Chat with Agent-Based Manager +Sample: Group Chat with Agent-Based Manager + Message Middleware What it does: -- Demonstrates the new set_manager() API for agent-based coordination -- Manager is a full Agent with access to tools, context, and observability -- Coordinates a researcher and writer agent to solve tasks collaboratively +- Demonstrates middleware integration for message interception between agents +- Manager coordinates Researcher and Writer collaboratively +- Middleware filters, tags, and controls message flow in real-time + +Middleware Use Cases & Patterns: +1. Content Filtering: Compliance, PII/secret prevention, toxicity control +2. Transformation: Audit tagging, format enforcement, prompt sanitization +3. Access Control: Multi-tenant routing, cost gates, agent permissioning +4. Ordering Rule: Access Control → Content Filter → Transformation → Logging +5. Failure Mode: Dropping messages (`None`) removes context → prefer redaction tokens +6. Integration Note: This sample uses a wrapper pattern. Native hooks (e.g., + .with_middleware()) can be added to GroupChatBuilder in future framework versions. Prerequisites: - FOUNDRY_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint. - FOUNDRY_MODEL must be set to your Azure OpenAI model deployment name. -- Authentication via azure-identity. Use AzureCliCredential and run az login before executing the sample. +- Authentication via azure-identity. Use AzureCliCredential and run az login before executing. """ ORCHESTRATOR_AGENT_INSTRUCTIONS = """ @@ -42,17 +153,12 @@ async def main() -> None: - # Create a Responses client using Azure OpenAI and Azure CLI credentials for all agents client = FoundryChatClient( project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], model=os.environ["FOUNDRY_MODEL"], credential=AzureCliCredential(), ) - # Orchestrator agent that manages the conversation - # Note: This agent (and the underlying chat client) must support structured outputs. - # The group chat workflow relies on this to parse the orchestrator's decisions. - # `response_format` is set internally by the GroupChat workflow when the agent is invoked. orchestrator_agent = Agent( name="Orchestrator", description="Coordinates multi-agent collaboration by selecting speakers", @@ -60,7 +166,6 @@ async def main() -> None: client=client, ) - # Participant agents researcher = Agent( name="Researcher", description="Collects relevant background information", @@ -75,34 +180,37 @@ async def main() -> None: client=client, ) + middleware_pipeline = compose_middleware( + create_content_filter(forbidden_terms=["confidential", "internal-ip", "password"]), + create_message_tagger(prefix="AUDIT"), + create_sender_blocklist(blocked_agents=set()) + ) + # Build the group chat workflow - # termination_condition: stop after 4 assistant messages - # (The agent orchestrator will intelligently decide when to end before this limit but just in case) - # Mark participant responses as intermediate so the stream shows the - # conversation as it unfolds while the orchestrator's transcript remains the - # terminal workflow output. + # TODO: Once native middleware support is added to GroupChatBuilder, + # replace the wrapper below with: .with_message_interceptor(middleware_pipeline) workflow = ( GroupChatBuilder( participants=[researcher, writer], - termination_condition=lambda messages: sum(1 for msg in messages if msg.role == "assistant") >= 4, intermediate_output_from=[researcher, writer], orchestrator_agent=orchestrator_agent, ) - # Set a hard termination condition: stop after 4 assistant messages - # The agent orchestrator will intelligently decide when to end before this limit but just in case .with_termination_condition(lambda messages: sum(1 for msg in messages if msg.role == "assistant") >= 4) .build() ) - task = "What are the key benefits of using async/await in Python? Provide a concise summary." + task = ( + "What are the key benefits of using async/await in Python? " + "Provide a concise summary. (Note: Do not include confidential or internal-ip details.)" + ) - print("\nStarting Group Chat with Agent-Based Manager...\n") + print("\nStarting Group Chat with Agent-Based Manager & Middleware...\n") print(f"TASK: {task}\n") print("=" * 80) - # Keep track of the last response to format output nicely in streaming mode + # Use the middleware wrapper instead of assuming a builder method last_response_id: str | None = None - async for event in workflow.run(task, stream=True): + async for event in run_with_message_middleware(workflow, task, middleware_pipeline, stream=True): if event.type in ("intermediate", "output"): data = event.data if isinstance(data, AgentResponseUpdate): @@ -114,13 +222,13 @@ async def main() -> None: last_response_id = rid print(data.text, end="", flush=True) elif event.type == "output": - # The output of the group chat workflow is a collection of chat messages from all participants - outputs = cast(list[Message], event.data) + outputs = cast(list[Message], data) print("\n" + "=" * 80) - print("\nFinal Conversation Transcript:\n") + print("\nFinal Conversation Transcript (after middleware):\n") for message in outputs: print(f"{message.author_name or message.role}: {message.text}\n") if __name__ == "__main__": asyncio.run(main()) + \ No newline at end of file From c2faf6a872729ec411cf3f7a100614f2a37bbb67 Mon Sep 17 00:00:00 2001 From: pratikwayase Date: Sat, 30 May 2026 15:50:02 +0530 Subject: [PATCH 2/2] fix: address middleware immutability, ordering, and Copilot review comments --- .../group_chat_agent_manager.py | 64 ++++++++++--------- 1 file changed, 35 insertions(+), 29 deletions(-) diff --git a/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py b/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py index 803c6eae9d..5ce158cedb 100644 --- a/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py +++ b/python/samples/03-workflows/orchestrations/group_chat_agent_manager.py @@ -1,6 +1,7 @@ # Copyright (c) Microsoft. All rights reserved. import asyncio +import dataclasses import os from typing import AsyncIterator, Callable, Optional, cast @@ -20,22 +21,22 @@ MessageInterceptor = Callable[[Message], Optional[Message]] - def compose_middleware(*middlewares: MessageInterceptor) -> MessageInterceptor: """Chain multiple interceptors. Execution stops early if any returns None.""" def chained_interceptor(msg: Message) -> Optional[Message]: current = msg for mw in middlewares: + current = mw(current) if current is None: break - current = mw(current) return current return chained_interceptor - # 1. CONTENT FILTERING: Drop or redact messages containing sensitive terms def create_content_filter(forbidden_terms: list[str], redact_instead: bool = True) -> MessageInterceptor: - def interceptor(msg: Message) -> Optional[Message]: + def interceptor(msg: Optional[Message]) -> Optional[Message]: + if msg is None: + return None if msg.role != "assistant" or not msg.text: return msg @@ -43,18 +44,17 @@ def interceptor(msg: Message) -> Optional[Message]: if any(term.lower() in text_lower for term in forbidden_terms): if redact_instead: safe_text = "[REDACTED BY MIDDLEWARE]" - try: - return msg.replace(text=safe_text) if hasattr(msg, "replace") else Message( - role=msg.role, author_name=msg.author_name, text=safe_text - ) - except Exception: - return None + if hasattr (msg,"replace"): + try : + return msg.replace(text = safe_text) + except TypeError: + pass + return Message (role = msg.role, author_name = msg.author_name, text = safe_text) print(f"\n[MIDDLEWARE] Content Filter: Dropped message from '{msg.author_name}'") return None return msg return interceptor - # 2. MESSAGE TRANSFORMATION: Add metadata prefixes for audit/tracing def create_message_tagger(prefix: str) -> MessageInterceptor: def interceptor(msg: Message) -> Optional[Message]: @@ -71,7 +71,6 @@ def interceptor(msg: Message) -> Optional[Message]: return msg return interceptor - # 3. ACCESS CONTROL: Restrict messages from specific agents def create_sender_blocklist(blocked_agents: set[str]) -> MessageInterceptor: def interceptor(msg: Message) -> Optional[Message]: @@ -89,36 +88,43 @@ async def run_with_message_middleware( *, stream: bool = True, ) -> AsyncIterator[Event]: - """ - Wraps workflow execution to apply message interception. - This is a user-space pattern until native builder hooks are added. - """ + async for event in workflow.run(task, stream=stream): + + # INTERMEDIATE EVENTS if event.type == "intermediate" and isinstance(event.data, AgentResponseUpdate): original_text = event.data.text + if original_text: temp_msg = Message( role=event.data.role or "assistant", author_name=event.data.author_name, text=original_text ) + intercepted = middleware(temp_msg) + if intercepted is None: - continue - event.data.text = intercepted.text + continue + + if intercepted.text != original_text: + new_data = dataclasses.replace( + event.data, + text=intercepted.text + ) + event = dataclasses.replace(event, data=new_data) + # OUTPUT EVENTS elif event.type == "output" and isinstance(event.data, list): filtered_messages = [] + for msg in event.data: if isinstance(msg, Message): processed = middleware(msg) if processed is not None: filtered_messages.append(processed) - event.data = filtered_messages - + event = dataclasses.replace(event, data=filtered_messages) yield event - - """ Sample: Group Chat with Agent-Based Manager + Message Middleware @@ -129,12 +135,12 @@ async def run_with_message_middleware( Middleware Use Cases & Patterns: 1. Content Filtering: Compliance, PII/secret prevention, toxicity control -2. Transformation: Audit tagging, format enforcement, prompt sanitization +2. Tagging/Transformation: Audit tagging, format enforcement, prompt sanitization 3. Access Control: Multi-tenant routing, cost gates, agent permissioning -4. Ordering Rule: Access Control → Content Filter → Transformation → Logging +4. Ordering in this sample: Content Filter → Tagger → Access Control 5. Failure Mode: Dropping messages (`None`) removes context → prefer redaction tokens 6. Integration Note: This sample uses a wrapper pattern. Native hooks (e.g., - .with_middleware()) can be added to GroupChatBuilder in future framework versions. +.with_middleware()) can be added to GroupChatBuilder in future framework versions. Prerequisites: - FOUNDRY_PROJECT_ENDPOINT must be your Azure AI Foundry Agent Service (V2) project endpoint. @@ -151,7 +157,6 @@ async def run_with_message_middleware( - Only finish after both have contributed meaningfully """ - async def main() -> None: client = FoundryChatClient( project_endpoint=os.environ["FOUNDRY_PROJECT_ENDPOINT"], @@ -183,7 +188,9 @@ async def main() -> None: middleware_pipeline = compose_middleware( create_content_filter(forbidden_terms=["confidential", "internal-ip", "password"]), create_message_tagger(prefix="AUDIT"), - create_sender_blocklist(blocked_agents=set()) + # Use a clearly non-existent agent name so this sample demonstrates + # sender blocking without affecting the current participants. + create_sender_blocklist(blocked_agents={"DemoBlockedAgent"}) ) # Build the group chat workflow @@ -230,5 +237,4 @@ async def main() -> None: if __name__ == "__main__": - asyncio.run(main()) - \ No newline at end of file + asyncio.run(main()) \ No newline at end of file