How do I design an event-driven messaging pattern for agents (pub-sub) so services can subscribe to agent events and scale independently?

Event-driven messaging is the difference between a toy agent demo and an agent platform that survives real traffic. If you want services to subscribe to agent events and scale independently, you need a pub-sub pattern that treats agents as event producers/consumers, not as addresses on a message bus. In AutoGen’s terms, that means using topics and subscriptions instead of hard‑coded agent IDs, and leaning on the runtime for routing and lifecycle.

Quick Answer: Design your agent messaging around topics and subscriptions, not direct agent IDs. In AutoGen Core, model events as Topic = (Topic Type, Topic Source), use TypeSubscription to map topic types to agent types, and let the runtime fan-out events to all interested agents/services. This gives you true pub-sub semantics, allows independent scaling of services, and keeps your agent graph portable across runtimes.

Why This Matters

Direct messaging between agents (“send this to agent X”) works in notebooks but breaks at scale. Once you have multiple services, tenants, and runtimes, you need a consistent way to:

• Broadcast agent events without knowing who will handle them.
• Add/remove consumers without modifying producers.
• Scale specific subscribers independently (e.g., heavy analytics vs. lightweight logging).

A topic‑based, event-driven pattern solves this by decoupling producers and consumers. In AutoGen Core, topics and subscriptions are runtime primitives, so you get deterministic routing, lifecycle management, and safety boundaries without inventing your own broker semantics.

Key Benefits:

• Loose coupling: Agents and services don’t need to know each other’s IDs; they only share topic conventions.
• Independent scaling: Heavy subscribers (e.g., analytics, observability, billing) can scale out without touching core agent logic.
• Portable topology: Using TypeSubscription and topic types keeps your design portable across runtimes (local SingleThreadedAgentRuntime vs. distributed worker topologies).

Core Concepts & Key Points

Concept	Definition	Why it's important
Topic	A routing key for messages defined as Topic = (Topic Type, Topic Source); string form Topic_Type/Topic_Source	Provides a stable channel for events so publishers don’t care who is listening
Subscription	A mapping from a topic (or topic type) to recipient agent IDs or agent types	Lets the runtime deliver each message to the right agents, handling one-to-many fan-out
Type-based Subscription (TypeSubscription)	A subscription that maps a topic type to an agent type without enumerating specific topics or agent IDs	Enables dynamic pub-sub, multi-tenant isolation, and adding new subscribers without changing producers

How It Works (Step-by-Step)

At a high level, designing an event-driven messaging pattern for agents (pub-sub) so services can subscribe to agent events and scale independently looks like this:

1. Define your event topics:

   • Decide what events agents will emit (e.g., task.completed, task.failed, audit.log).
   • For each event, define:
     • topic_type: e.g., "task.completed", "task.failed", "audit.log".
     • topic_source: e.g., tenant ID, workflow ID, or agent ID (for multi-tenant isolation).
   • Use the Topic_Type/Topic_Source string convention in your design docs so it’s unambiguous.

2. Configure subscriptions (including TypeSubscription):

   • Use type-based subscriptions to map topic types to agent types:
     • TypeSubscription(topic_type="task.completed", agent_type="task_sink_agent")
     • This creates an unbounded mapping: any task.completed topic is routed to all task_sink_agent instances.
   • For single-tenant, multi-topic scenarios, create one type-based subscription per specialization so different agent types handle different topics.
   • For shared topics, map the same topic type to multiple agent types when several services should consume the same events.

3. Let agents publish events instead of targeting IDs:

   • When an agent finishes work, it publishes to the appropriate topic rather than sending to a specific agent ID.
   • The runtime uses subscriptions to deliver the message to all matching recipients, ensuring each recipient gets it only once.
   • Services and downstream agents scale by adding more instances of the subscribed agent types, not by changing publishers.

---

Below, I’ll walk through a minimal AutoGen Core setup that demonstrates this pattern, then zoom into common mistakes and a concrete example.

Note: Python 3.10 or later is required for AutoGen. The code samples assume you’re comfortable with basic Python and async patterns.

Installation

For event-driven patterns at the runtime layer, you want AutoGen Core plus Extensions for model clients/tools:

pip install -U "autogen-core" "autogen-ext[openai]"

If you also want to prototype behaviors quickly with higher-level abstractions, add AgentChat:

pip install -U "autogen-agentchat"

For this article, the focus is on Core’s topics/subscriptions, but you can wrap the same patterns around AgentChat agents.

Designing the Pub-Sub Pattern with Topics & Subscriptions

Step 1: Define Topics

In AutoGen Core, you should think in terms of:

• Topic = (Topic Type, Topic Source)
• String form: "{topic_type}/{topic_source}"

Examples:

• task.completed/tenant_123
• task.failed/tenant_123
• audit.log/tenant_123
• agent.metrics/workflow_42

Guidelines:

• Use topic type to encode what happened (semantic event).
• Use topic source to encode where it happened (tenant/workflow/agent), which gives you isolation and routing control.

This is how you get event-driven multi-tenancy without hard-coding agent IDs: the same topic type can exist across many topic sources.

Step 2: Configure Subscriptions

There are two subscription layers you should care about:

1. Direct subscription (topic → agent ID):
   Good for tightly controlled workflows or one-off integrations.

2. Type-based subscription (TypeSubscription) (topic type → agent type):
   This is your core pub-sub primitive at scale.

From the docs:

• A subscription maps a topic to agent IDs.
• If a topic has no subscription, messages published to this topic will not be delivered.
• If a topic has many subscriptions, messages will be delivered following all the subscriptions to every recipient agent only once.
• A type-based subscription maps a topic type to an agent type. It declares an unbounded mapping from topics to agent IDs without knowing the exact topic sources and agent keys.

That last line is the key for “services can subscribe to agent events and scale independently”:

• You don’t enumerate every tenant_id/workflow_id.
• You say: “For any task.completed event, route it to all agents of type task_sink_agent.”
• When new tenants or workflows appear, they automatically join the pub-sub pattern as long as they use the same topic type.

Step 3: Implement an Event-Emitting Agent

Assume you have a worker agent that processes tasks and emits a completion event.

Pseudo-structure:

• Input topic (e.g., task.request/tenant_123)
• Output event topic (e.g., task.completed/tenant_123)

Your agent logic should:

1. Handle an incoming message from a “request” topic.
2. Perform the work (could call an LLM via OpenAIChatCompletionClient).
3. Publish a new message to the task.completed topic using the same topic_source for isolation.

In a SingleThreadedAgentRuntime, this looks roughly like:

import asyncio
from autogen_core.base import Agent, Message
from autogen_core.runtime import SingleThreadedAgentRuntime

class TaskWorkerAgent(Agent):
    async def on_receive(self, message: Message, runtime: SingleThreadedAgentRuntime):
        # Do some work – placeholder
        result = {"status": "done", "input": message.content}

        # Derive topic_source from the incoming message (e.g., tenant_id)
        topic_source = message.metadata.get("tenant_id", "default")

        # Publish a completion event
        await runtime.publish(
            topic_type="task.completed",
            topic_source=topic_source,
            content=result,
        )

Here the agent never references other agent IDs. It just publishes to a topic; the runtime handles who gets it.

Step 4: Implement Subscriber Agents/Services via TypeSubscription

Now define one or more subscriber agent types that consume task.completed:

class TaskLoggerAgent(Agent):
    async def on_receive(self, message: Message, runtime: SingleThreadedAgentRuntime):
        print(f"[LOG] Completed task for tenant={message.metadata.get('tenant_id')}: {message.content}")

class TaskAnalyticsAgent(Agent):
    async def on_receive(self, message: Message, runtime: SingleThreadedAgentRuntime):
        # send to your metrics backend, data warehouse, etc.
        print(f"[ANALYTICS] Emitting metrics for task: {message.content}")

Configure type-based subscriptions so any task.completed topic goes to both agent types:

async def main():
    runtime = SingleThreadedAgentRuntime()

    # Register agent types
    runtime.register_agent_type("task_worker", TaskWorkerAgent)
    runtime.register_agent_type("task_logger", TaskLoggerAgent)
    runtime.register_agent_type("task_analytics", TaskAnalyticsAgent)

    # Type-based subscriptions: topic_type -> agent_type
    await runtime.add_type_subscription(
        topic_type="task.completed",
        agent_type="task_logger",
    )
    await runtime.add_type_subscription(
        topic_type="task.completed",
        agent_type="task_analytics",
    )

    # Also subscribe a worker to task.request events
    await runtime.add_type_subscription(
        topic_type="task.request",
        agent_type="task_worker",
    )

    # Publish a request event
    await runtime.publish(
        topic_type="task.request",
        topic_source="tenant_123",
        content={"task_id": "T-1", "payload": "do something"},
        metadata={"tenant_id": "tenant_123"},
    )

    await runtime.run()  # pump events until completion or external stop

if __name__ == "__main__":
    asyncio.run(main())

What happens here:

• The runtime receives task.request/tenant_123.
• Via TypeSubscription(topic_type="task.request", agent_type="task_worker"), it creates (if needed) and invokes a TaskWorkerAgent.
• The worker completes the task and publishes to task.completed/tenant_123.
• The runtime sees two type-based subscriptions for task.completed:
  • task_logger and task_analytics.
• It delivers the event to both agent types, and each agent instance sees it once.

This is the core event-driven pub-sub pattern with agents that can scale independently.

Step 5: Scale Out with a Distributed Runtime

When you outgrow a single process, you move to a distributed runtime (host servicer + workers + gateways) but keep the same topic/TypeSubscription design. You still publish to topic types, and your subscriber agent types can now be hosted across many workers.

In that topology:

• Gateways receive publish calls.
• The host maintains the subscription map (topic types → agent types/IDs).
• Workers host agent instances that consume messages.

You can scale a specific subscriber (e.g., analytics) by:

• Registering TaskAnalyticsAgent on more workers.
• Keeping the same type-based subscription for task.completed.

The engineering win: your design doesn’t change as you scale the runtime; you only change deployment topology.

Note: Always think of topics and subscriptions as runtime contracts, not code-level shortcuts. This makes moving from SingleThreadedAgentRuntime to a distributed runtime almost mechanical.

Common Mistakes to Avoid

• Hard-coding agent IDs instead of using topics:

  • This couples producers to consumers and makes scaling or refactoring painful.
  • How to avoid it: Always publish to topic_type/topic_source and let the runtime handle agent creation and delivery via TypeSubscription.

• Using topic source as a random string with no isolation semantics:

  • If topic_source doesn’t encode something meaningful (tenant, workflow, or agent), you lose the ability to enforce multi-tenant isolation or to shard workloads cleanly.
  • How to avoid it: Design a clear format for topic_source (e.g., tenant_id or tenant_id:workflow_id) and document it for your team.

• One “catch-all” topic for everything:

  • This turns your event bus into a noisy firehose and forces subscribers to do heavy filtering in code.
  • How to avoid it: Define specific topic types per event category (task.completed, task.failed, audit.log, agent.metrics) so subscriptions stay targeted.

• Not leveraging type-based subscription for new use cases:

  • Teams often add new direct subscribers, creating brittle point-to-point links.
  • How to avoid it: When a new service wants to consume an event, add a TypeSubscription for its agent type; don’t modify the producer.

• Ignoring unsubscribe/lifecycle:

  • Stale subscriptions can lead to ghost consumers or unexpected fan-out.
  • How to avoid it: Treat subscriptions as configuration; manage them via runtime APIs and automate cleanup when decommissioning services.

Real-World Example

Here’s how we use this pattern internally in a regulated environment where different services must observe agent events but scale independently.

Scenario

• A multi-tenant agent platform processes user tasks.
• Each tenant’s tasks run as workflows orchestrated by agents.
• Compliance requires:
  • An immutable audit log of every completed task.
  • A separate analytics pipeline for metrics and anomaly detection.
  • Isolation between tenants.

Design

1. Topics:

   • task.completed/{tenant_id}
   • task.failed/{tenant_id}
   • audit.log/{tenant_id}

2. Producer:

   • A TaskOrchestratorAgent publishes:
     • task.completed/tenant_123 on success.
     • task.failed/tenant_123 on failure.

3. Subscribers:

   • AuditSinkAgent (audit service)
   • MetricsSinkAgent (analytics service)
   • Both are implemented as agents and deployed across a distributed runtime.

4. Type-based subscriptions:

   • TypeSubscription(topic_type="task.completed", agent_type="audit_sink_agent")
   • TypeSubscription(topic_type="task.failed", agent_type="audit_sink_agent")
   • TypeSubscription(topic_type="task.completed", agent_type="metrics_sink_agent")

We never add code in TaskOrchestratorAgent to contact audit or metrics directly. As new services (e.g., billing) appear, they simply subscribe to task.completed via a new TypeSubscription.

Operationally, this yields:

• Strong isolation: Tenant ID lives in topic_source, so we can route and shard by tenant.
• Independent scaling: If analytics needs more capacity, we add workers that host MetricsSinkAgent only.
• Observability: Adding a DebugTapAgent to task.* topics is a one-line subscription change, not an agent code change.

Pro Tip: Treat topics and subscriptions as part of your platform configuration (e.g., Terraform or a “routing config” repo). Code focuses on what events to publish; configuration decides who listens. This keeps your event-driven agent graph adaptable as new services come and go.

Summary

To design an event-driven messaging pattern for agents (pub-sub) so services can subscribe to agent events and scale independently, don’t start with agent IDs—start with topics. In AutoGen Core, use Topic = (Topic Type, Topic Source) as the primary routing primitive and configure TypeSubscription mappings from topic types to agent types. Agents publish events to topics; the runtime performs fan-out so subscribers can be added, scaled, or removed without changing producers.

This pattern transfers cleanly from SingleThreadedAgentRuntime to distributed runtimes and works well alongside higher-level abstractions in AgentChat or GraphFlow. It also gives you the engineering controls you care about in production: predictable routing, multi-tenant isolation, and the ability to add observability/analytics/billing as independent subscribers.

Next Step

Get Started