How MsgHub Enables Multi-Agent Communication and Message Routing in Agentscope

MsgHub acts as a central coordination primitive that automatically routes agent replies to all other registered participants by maintaining subscriber lists within each AgentBase instance.

The MsgHub class in the agentscope-ai/agentscope repository provides the backbone for multi-agent communication and message routing, eliminating the need for agents to maintain explicit references to one another. By leveraging asynchronous context managers and an internal subscriber registry, MsgHub orchestrates message propagation across agent groups with minimal boilerplate code.

Core Architecture and Subscriber Management

The Subscriber Map Implementation

At the heart of MsgHub's routing capability lies the _subscribers dictionary defined in src/agentscope/agent/_agent_base.py. This private attribute stores a mapping where keys are unique hub names (generated via shortuuid) and values are lists of agent instances that should receive messages from the owning agent.

When a MsgHub instance enters its asynchronous context via __aenter__ in src/agentscope/pipeline/_msghub.py, it invokes _reset_subscriber(), which calls AgentBase.reset_subscribers(hub_name, participants) for every participant. This method populates each agent's _subscribers entry with all other agents in the hub, excluding the agent itself.

Context Manager Lifecycle

The MsgHub implements __aenter__ and __aexit__ to guarantee clean setup and teardown of routing tables. Upon entry, the hub generates a unique identifier and optionally broadcasts a startup announcement. Upon exit, remove_subscribers cleans up the _subscribers entries to prevent memory leaks and stale references.

Message Routing Execution Flow

The routing mechanism operates through a tightly integrated flow between the MsgHub coordinator and individual agent instances:

1. Agent Invocation: When an agent is called via await agent(), the AgentBase.__call__ method in src/agentscope/agent/_agent_base.py executes the agent's reply logic and subsequently triggers self._broadcast_to_subscribers(reply_msg).

2. Subscriber Iteration: The _broadcast_to_subscribers method iterates over the agent's _subscribers dictionary, locating the list associated with the current hub context.

3. Message Delivery: For each subscriber in the list, the method invokes subscriber.observe(msg), delivering the message to the recipient agent's observation handler.

This architecture ensures that every reply generated within a hub context automatically reaches all other participants without explicit send calls from the developer.

Broadcast Modes and Configuration

Automatic Broadcasting

By default, MsgHub operates with enable_auto_broadcast=True, creating a fully connected message mesh where every agent reply propagates to all other registered agents immediately upon generation. This mode suits collaborative scenarios requiring total information visibility across the agent group.

Manual Broadcasting

When enable_auto_broadcast=False, the hub suppresses automatic message propagation. Developers must explicitly invoke hub.broadcast(msg) to distribute messages to all participants. This mode provides fine-grained control over information disclosure and reduces network overhead in selective communication scenarios.

Dynamic Participant Management

Runtime Agent Addition

The add(new_agent) method in src/agentscope/pipeline/_msghub.py enables dynamic expansion of the agent group. Upon invocation, the method appends the new agent to the participants list and triggers _reset_subscriber() to rebuild the routing tables, ensuring existing agents receive the new participant's messages and vice versa.

Agent Removal and Cleanup

Conversely, delete(old_agent) removes an agent from the participants list and reinitializes the subscriber mappings. When the hub context exits or auto-broadcast is disabled, the __aexit__ method ensures all participants remove the hub's entry from their _subscribers dictionaries via remove_subscribers.

Practical Implementation Example

The following example demonstrates both automatic and manual routing modes using the MsgHub API:

from agentscope.agent import AgentBase
from agentscope.message import Msg
from agentscope.pipeline import MsgHub
import asyncio

class EchoAgent(AgentBase):
    async def reply(self, *_, **__) -> Msg:
        return Msg(self.name, f"hello from {self.name}", "assistant")

    async def observe(self, msg: Msg) -> None:
        print(f"[{self.name} observed] {msg.content}")

    async def handle_interrupt(self, *_, **__) -> Msg:
        pass

async def demo():
    a = EchoAgent(); a.name = "A"
    b = EchoAgent(); b.name = "B"
    c = EchoAgent(); c.name = "C"

    # Automatic broadcast mode

    async with MsgHub(participants=[a, b, c]) as hub:
        await a()  # A's reply automatically reaches B and C

        await b()  # B's reply automatically reaches A and C

    # Manual broadcast mode

    hub = MsgHub(participants=[a, b, c], enable_auto_broadcast=False)
    async with hub:
        await hub.broadcast(Msg("system", "manual round", "assistant"))
        msg = await a()
        await hub.broadcast(msg)  # Explicit broadcast required

asyncio.run(demo())

In the automatic mode, invoking await a() triggers the internal _broadcast_to_subscribers method, which delivers the message to b.observe() and c.observe() without additional code.

Summary

• MsgHub coordinates multi-agent groups by managing per-agent subscriber lists in AgentBase._subscribers.
• Automatic routing occurs via _broadcast_to_subscribers inside AgentBase.__call__, forwarding every reply to all other participants.
• Context manager semantics ensure routing tables initialize on entry and clean up on exit through __aenter__ and __aexit__ in src/agentscope/pipeline/_msghub.py.
• Dynamic modifications are supported through add() and delete() methods that rebuild subscriber mappings via _reset_subscriber.
• Broadcast control via enable_auto_broadcast toggles between automatic mesh networking and explicit multicast patterns.

Frequently Asked Questions

How does MsgHub differ from direct agent-to-agent messaging?

MsgHub eliminates the need for agents to maintain explicit references to communication partners. Instead of agents calling each other directly, they register with a MsgHub instance that manages the _subscribers registry in src/agentscope/agent/_agent_base.py, automatically routing messages through the observe() method of all registered participants.

What happens to message routing when an agent is added mid-execution?

When hub.add(new_agent) is called, the MsgHub rebuilds the subscriber mappings by invoking _reset_subscriber(), which updates the _subscribers dictionary for every participant in src/agentscope/pipeline/_msghub.py. This ensures the new agent receives future messages from existing agents and that its replies reach the established group.

Can MsgHub operate without automatic message broadcasting?

Yes. Setting enable_auto_broadcast=False in the MsgHub constructor disables automatic propagation. In this mode, developers must explicitly call hub.broadcast(msg) to distribute messages, providing manual control over the multi-agent communication flow while still leveraging the hub's subscriber management infrastructure.

How does MsgHub prevent memory leaks when agents leave the conversation?

The MsgHub uses Python's asynchronous context manager protocol. When exiting the async with block, __aexit__ invokes remove_subscribers on each participant, deleting the hub's entry from their _subscribers dictionaries in src/agentscope/agent/_agent_base.py and ensuring no stale references persist after the communication session ends.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}