Implementation:Microsoft Autogen MessageFilterAgent Init

Knowledge Sources	Microsoft AutoGen
Domains	Multi-Agent Systems, Message Filtering, Context Management, Information Scoping
Last Updated	2026-02-11 00:00 GMT

Overview

Concrete tool for filtering incoming messages before they reach an inner agent, controlling per-source message context in graph workflows, provided by Microsoft AutoGen.

Description

MessageFilterAgent is a wrapper agent that intercepts the incoming message sequence, applies source-based filters to select a relevant subset, and then delegates processing to the wrapped inner agent. The inner agent receives only the filtered messages and is unaware that any filtering occurred.

Filtering is configured via MessageFilterConfig, which contains a list of PerSourceFilter entries. Each entry specifies a source name and optionally a positional selector ("first" or "last") with a count. Messages from sources not listed in any filter are excluded entirely.

The agent is compatible with both direct message passing (on_messages) and streaming execution (on_messages_stream), and works seamlessly within GraphFlow team-based workflows.

Usage

Use MessageFilterAgent when agents in a graph-based workflow should only see a subset of the conversation history. Common scenarios include:

• Wrapping agents in iterative loops so they see only the latest feedback, not the full loop history.
• Ensuring an agent only sees the original user request and the most recent output from its predecessor.
• Allowing an agent to see a bounded window of its own prior responses for self-reflection.

Code Reference

Source Location

• Repository: Microsoft AutoGen
• File: python/packages/autogen-agentchat/src/autogen_agentchat/agents/_message_filter_agent.py (Lines 140-148)

Signature

class MessageFilterAgent(BaseChatAgent, Component[MessageFilterAgentConfig]):
    def __init__(
        self,
        name: str,
        wrapped_agent: BaseChatAgent,
        filter: MessageFilterConfig,
    ):

class MessageFilterConfig(BaseModel):
    per_source: List[PerSourceFilter]

class PerSourceFilter(BaseModel):
    source: str
    position: Optional[Literal["first", "last"]] = None
    count: Optional[int] = None

Import

from autogen_agentchat.agents import MessageFilterAgent
from autogen_agentchat.agents._message_filter_agent import (
    MessageFilterConfig,
    PerSourceFilter,
)

I/O Contract

Inputs

Name	Type	Required	Description
name	str	Yes	The name for this wrapper agent. This is the name used in the graph node and message source attribution.
wrapped_agent	BaseChatAgent	Yes	The inner agent that will receive filtered messages. Can be any agent implementing the BaseChatAgent protocol (e.g., AssistantAgent).
filter	MessageFilterConfig	Yes	Configuration specifying per-source message filtering rules. Contains a list of PerSourceFilter entries.

PerSourceFilter Fields

Name	Type	Required	Description
source	str	Yes	The message source to match (e.g., "user", another agent's name).
position	Literal["first", "last"] or None	No	If "first", selects from the beginning of the source's messages. If "last", selects from the end. If None, all messages from the source are included.
count	int or None	No	Number of messages to include from the selected position. Only meaningful when position is set.

Outputs

Name	Type	Description
MessageFilterAgent	MessageFilterAgent	A wrapper agent that filters messages before delegating to the inner agent. Produces the same message types as the wrapped agent.

Usage Examples

Basic Example: Filter to Last Message from Each Source

from autogen_agentchat.agents import AssistantAgent, MessageFilterAgent
from autogen_agentchat.agents._message_filter_agent import (
    MessageFilterConfig,
    PerSourceFilter,
)
from autogen_ext.models.openai import OpenAIChatCompletionClient

model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

# Create the inner agent
inner_agent = AssistantAgent(
    "translator",
    model_client=model_client,
    system_message="Translate the input to French.",
)

# Wrap with message filtering: see only user's first message and last message from agent B
filtered_agent = MessageFilterAgent(
    name="translator",
    wrapped_agent=inner_agent,
    filter=MessageFilterConfig(
        per_source=[
            PerSourceFilter(source="user", position="first", count=1),
            PerSourceFilter(source="B", position="last", count=1),
        ]
    ),
)

Graph Workflow Example: Loop with Filtered Context

import asyncio
from autogen_agentchat.agents import AssistantAgent, MessageFilterAgent
from autogen_agentchat.agents._message_filter_agent import (
    MessageFilterConfig,
    PerSourceFilter,
)
from autogen_agentchat.conditions import MaxMessageTermination
from autogen_agentchat.teams import DiGraphBuilder, GraphFlow
from autogen_ext.models.openai import OpenAIChatCompletionClient


async def main():
    model_client = OpenAIChatCompletionClient(model="gpt-4.1-nano")

    # Create inner agents
    writer_inner = AssistantAgent("writer", model_client=model_client,
                                  system_message="Write a short poem.")
    reviewer_inner = AssistantAgent("reviewer", model_client=model_client,
                                    system_message="Review the poem. Say APPROVE if good.")

    # Wrap with filters: writer sees user + last review; reviewer sees user + last poem
    writer = MessageFilterAgent(
        name="writer",
        wrapped_agent=writer_inner,
        filter=MessageFilterConfig(per_source=[
            PerSourceFilter(source="user", position="first", count=1),
            PerSourceFilter(source="reviewer", position="last", count=1),
        ]),
    )
    reviewer = MessageFilterAgent(
        name="reviewer",
        wrapped_agent=reviewer_inner,
        filter=MessageFilterConfig(per_source=[
            PerSourceFilter(source="user", position="first", count=1),
            PerSourceFilter(source="writer", position="last", count=1),
        ]),
    )

    # Build graph with a review loop
    builder = DiGraphBuilder()
    builder.add_node(writer).add_node(reviewer)
    builder.add_edge(writer, reviewer)
    builder.add_edge(reviewer, writer,
                     condition=lambda msg: "APPROVE" not in msg.to_model_text())
    builder.set_entry_point(writer)
    graph = builder.build()

    team = GraphFlow(
        participants=[writer, reviewer],
        graph=graph,
        termination_condition=MaxMessageTermination(10),
    )

    async for event in team.run_stream(task="Write a haiku about autumn."):
        print(event)


asyncio.run(main())

Related Pages

Implements Principle

• Principle:Microsoft_Autogen_Message_Filtering
• Environment:Microsoft_Autogen_Python_Runtime_Environment