# Implementing Message Filtering with MessageFilterAgent and PerSourceFilter in AutoGen > Master AutoGen message filtering with MessageFilterAgent and PerSourceFilter. Control message access to inner agents, ensuring efficient communication and focused context. Learn how to implement advanced filtering for your AI a... - Repository: [Microsoft/autogen](https://github.com/microsoft/autogen) - Tags: how-to-guide - Published: 2026-03-07 --- **Use `MessageFilterAgent` to wrap any `BaseChatAgent` and apply `PerSourceFilter` rules that expose only specific subsets of messages—such as the last N messages from a particular source—before forwarding them to the inner agent.** Implementing message filtering with `MessageFilterAgent` and `PerSourceFilter` gives you surgical control over conversational context in the `microsoft/autogen` framework. This lightweight wrapper intercepts incoming chat history, applies configurable per-source rules, and passes only the filtered message slice to the wrapped agent. It is essential for managing context windows in looping multi-agent graphs and hiding irrelevant traffic from downstream participants. ## Understanding the MessageFilterAgent Architecture The filtering system consists of two primary layers: the data models that define *what* to filter, and the wrapper agent that executes the filtering logic. ### Core Data Models: PerSourceFilter and MessageFilterConfig At the heart of the implementation are Pydantic models defined in [`python/packages/autogen-agentchat/src/autogen_agentchat/agents/_message_filter_agent.py`](https://github.com/microsoft/autogen/blob/main/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_message_filter_agent.py). **`PerSourceFilter`** defines a single rule: ```python class PerSourceFilter(BaseModel): source: str # e.g., "user", "system", "AgentA" position: Optional[Literal["first", "last"]] = None count: Optional[int] = None ``` - **`source`**: The message origin to match. - **`position`**: Slice from the `"first"` or `"last"` of the matched set; `None` keeps all. - **`count`**: How many messages to include when `position` is specified. **`MessageFilterConfig`** aggregates rules: ```python class MessageFilterConfig(BaseModel): per_source: List[PerSourceFilter] # Empty list blocks all messages ``` An empty `per_source` list results in **all messages being blocked**, which is useful for creating silent agents that only respond to specific triggers. ### The Filtering Algorithm The `_apply_filter` method implements the selection logic: ```python def _apply_filter(self, messages: Sequence[BaseChatMessage]) -> Sequence[BaseChatMessage]: result: List[BaseChatMessage] = [] for source_filter in self._filter.per_source: msgs = [m for m in messages if m.source == source_filter.source] if source_filter.position == "first" and source_filter.count: msgs = msgs[: source_filter.count] elif source_filter.position == "last" and source_filter.count: msgs = msgs[-source_filter.count :] result.extend(msgs) return result ``` The algorithm preserves the original message order as it iterates through the configured rules, appending matched subsets to the result list. This ensures that downstream agents receive a coherent, chronologically consistent view of the conversation even when filtering is aggressive. ## Implementing Message Filtering in Practice ### Basic Usage with a Wrapped Agent To implement message filtering, wrap any `BaseChatAgent` with `MessageFilterAgent` and supply a `MessageFilterConfig`. The following example, based on the test suite in [`python/packages/autogen-agentchat/tests/test_group_chat_graph.py`](https://github.com/microsoft/autogen/blob/main/python/packages/autogen-agentchat/tests/test_group_chat_graph.py), demonstrates filtering for the last user message and the first system message: ```python from autogen_agentchat.agents import MessageFilterAgent, MessageFilterConfig, PerSourceFilter from autogen_agentchat.messages import TextMessage from autogen_core import CancellationToken # Minimal inner agent that records received messages class EchoAgent(BaseChatAgent): def __init__(self, name: str): super().__init__(name=name, description="Echo") self.received_messages = [] async def on_messages(self, messages, token: CancellationToken): self.received_messages.extend(messages) return Response() # dummy response inner = EchoAgent("inner") # Wrap with filtering rules wrapper = MessageFilterAgent( name="filtered", wrapped_agent=inner, filter=MessageFilterConfig( per_source=[ PerSourceFilter(source="user", position="last", count=1), PerSourceFilter(source="system", position="first", count=1) ] ), ) # Simulate incoming messages msgs = [ TextMessage(source="user", content="first user"), TextMessage(source="user", content="second user"), TextMessage(source="system", content="system init"), TextMessage(source="system", content="system follow-up"), ] await wrapper.on_messages(msgs, CancellationToken()) print([m.content for m in inner.received_messages]) # Output: ['second user', 'system init'] ``` The `EchoAgent` only receives `'second user'` (the last user message) and `'system init'` (the first system message), demonstrating how `MessageFilterAgent` shields the inner agent from full context. ### Integrating Filters in Graph-Based Workflows In complex multi-agent graphs, `MessageFilterAgent` prevents context explosion in loops. When agents pass messages cyclically, unfiltered history can grow indefinitely. By wrapping each node with specific `PerSourceFilter` rules, you ensure each agent sees only the messages necessary for its decision. The following pattern, derived from [`test_group_chat_graph.py`](https://github.com/microsoft/autogen/blob/main/test_group_chat_graph.py), shows how to configure filters for a three-agent graph where `A` and `B` loop, and `B` eventually exits to `C`: ```python from autogen_agentchat.teams import DiGraph, DiGraphNode, DiGraphEdge # Agent A: sees only the first user message and the last message from B agent_a = MessageFilterAgent( name="A", wrapped_agent=A_inner, filter=MessageFilterConfig( per_source=[ PerSourceFilter(source="user", position="first", count=1), PerSourceFilter(source="B", position="last", count=1), ] ), ) # Agent B: sees first user, last from A, and last 10 from itself (for context) agent_b = MessageFilterAgent( name="B", wrapped_agent=B_inner, filter=MessageFilterConfig( per_source=[ PerSourceFilter(source="user", position="first", count=1), PerSourceFilter(source="A", position="last", count=1), PerSourceFilter(source="B", position="last", count=10), ] ), ) # Agent C: sees first user and last from B agent_c = MessageFilterAgent( name="C", wrapped_agent=C_inner, filter=MessageFilterConfig( per_source=[ PerSourceFilter(source="user", position="first", count=1), PerSourceFilter(source="B", position="last", count=1), ] ), ) graph = DiGraph( nodes={ "A": DiGraphNode(name="A", edges=[DiGraphEdge(target="B")]), "B": DiGraphNode( name="B", edges=[ DiGraphEdge(target="C", condition="exit"), DiGraphEdge(target="A", condition="loop"), ], ), "C": DiGraphNode(name="C", edges=[]), }, default_start_node="A", ) ``` This configuration ensures that even if agents `A` and `B` loop dozens of times, each invocation receives a bounded, relevant context rather than the full, ever-growing message history. ### Serialization and Component Reuse `MessageFilterAgent` fully supports AutoGen’s component model for persistence and graph reconstruction. You can serialize the wrapper and its inner agent, then reload them later without losing configuration. ```python # Dump to a component model (e.g., for persistence or graph reconstruction) config = wrapper.dump_component() # Later… reload the exact same wrapper (including the inner agent) restored = MessageFilterAgent.load_component(config) assert restored.name == wrapper.name assert restored._filter == wrapper._filter ``` The `dump_component()` method returns a `MessageFilterAgentConfig` containing the wrapper’s name, the wrapped agent’s serialized component model, and the filter configuration. This enables version-controlled agent definitions and reproducible multi-agent deployments. ## Key Source Files and Implementation Details The implementation resides in the `autogen-agentchat` package. Refer to these files for the exact method signatures and internal logic: | File | Contents | |------|----------| | [`python/packages/autogen-agentchat/src/autogen_agentchat/agents/_message_filter_agent.py`](https://github.com/microsoft/autogen/blob/main/python/packages/autogen-agentchat/src/autogen_agentchat/agents/_message_filter_agent.py) | Defines `PerSourceFilter`, `MessageFilterConfig`, `MessageFilterAgentConfig`, and the `MessageFilterAgent` class with `_apply_filter`, `on_messages`, and component serialization methods. | | [`python/packages/autogen-agentchat/src/autogen_agentchat/agents/__init__.py`](https://github.com/microsoft/autogen/blob/main/python/packages/autogen-agentchat/src/autogen_agentchat/agents/__init__.py) | Public API exports for `MessageFilterAgent`, `MessageFilterConfig`, and `PerSourceFilter`. | | [`python/packages/autogen-agentchat/tests/test_group_chat_graph.py`](https://github.com/microsoft/autogen/blob/main/python/packages/autogen-agentchat/tests/test_group_chat_graph.py) | Unit tests verifying empty filter behavior, `position=None` handling, and integration within `DiGraph` workflows. | ## Summary - **`MessageFilterAgent`** wraps any `BaseChatAgent` to intercept and prune message history before invocation. - **`PerSourceFilter`** rules specify which sources to include, with optional `position` (`first`/`last`) and `count` constraints. - An empty `MessageFilterConfig` blocks all messages, useful for creating conditional or silent agents. - The wrapper supports full component serialization via `dump_component()` and `load_component()`, enabling reusable, version-controlled agent graphs. - Integration with `DiGraph` workflows prevents context explosion in cyclic multi-agent topologies by ensuring each node receives only relevant, bounded history. ## Frequently Asked Questions ### How does MessageFilterAgent handle messages when position is set to None? When `position` is `None` in a `PerSourceFilter`, the filter includes **all** messages from the specified source regardless of their position in the history. The `count` parameter is ignored in this case because the filter does not slice the result. This is useful when you want an agent to see the complete conversation from a specific participant while still filtering out noise from others. ### Can I use MessageFilterAgent to block all incoming messages? Yes. To block all messages, initialize `MessageFilterAgent` with a `MessageFilterConfig` that has an empty `per_source` list (`per_source=[]`). When the filter configuration contains no rules, the `_apply_filter` method returns an empty list, ensuring the wrapped agent receives no context. This pattern is useful for creating agents that respond only to internal state or specific external triggers rather than conversational history. ### Does MessageFilterAgent support streaming responses? Yes. `MessageFilterAgent` implements both `on_messages` and `on_messages_stream`. In the streaming case, it first applies the filter to the incoming message history using `_apply_filter`, then delegates to the wrapped agent's `on_messages_stream` method with the pruned context. The streaming iterator yields tokens from the inner agent while maintaining the filtered view of history, ensuring consistent context management regardless of whether you use synchronous or asynchronous streaming patterns. ### How do I persist and reload a MessageFilterAgent configuration? Use the component model serialization methods. Call `dump_component()` on a `MessageFilterAgent` instance to obtain a `MessageFilterAgentConfig` containing the wrapper's name, the serialized inner agent, and the filter rules. Later, pass this config to `MessageFilterAgent.load_component()` to reconstruct the exact same wrapper with its inner agent and filtering rules intact. This enables version-controlled agent definitions and reproducible deployment in production graphs.