Apache Kafka를 AI 에이전트 이벤트 브로커로 — MCP·A2A 멀티에이전트 시스템 대규모 확장 패턴

# HTTP 기반 (포인트-투-포인트)
에이전트A → 에이전트B
에이전트A → 에이전트C
에이전트B → 에이전트C
에이전트B → 에이전트D
...
# n개 에이전트 = 최대 n*(n-1)/2개 직접 연결
# 한 에이전트 장애 = 연결된 모든 에이전트에 전파
 
# Kafka 기반 (발행-구독)
에이전트A → [Kafka 토픽: price-events] → 에이전트B, C, D (구독)
에이전트B → [Kafka 토픽: analysis-results] → 에이전트E, F (구독)
# 에이전트는 토픽만 알면 됨. 서로를 몰라도 됨
# 한 에이전트 장애 = 해당 에이전트만 영향

┌──────────────────────────────────────────────────────┐
│                     Kafka Topics                      │
│                                                       │
│  [task-requests]  [task-results]  [dead-letter]       │
└──────────┬──────────────┬─────────────────────────────┘
           │              │
    ┌──────┴──────┐       │
    │ Orchestrator│◄──────┘
    │   Agent     │  결과 수집 및
    │ (A2A 위임)  │  다음 단계 결정
    └──────┬──────┘
           │ 작업 발행
    ┌──────┴──────────────────────┐
    │                             │
┌───┴────┐  ┌────────┐  ┌────────┴┐
│Worker A│  │Worker B│  │Worker C │
│(분석)  │  │(가격)  │  │(알림)   │
└────────┘  └────────┘  └─────────┘

{
  "name": "PriceAnalysisAgent",
  "description": "경쟁사 가격 데이터를 분석해 가격 조정 권고를 생성합니다",
  "url": "http://price-analysis-agent/",
  "capabilities": {
    "streaming": false,
    "pushNotifications": true
  },
  "skills": [
    {
      "id": "analyze-price-gap",
      "name": "가격 차이 분석",
      "description": "제품 ID와 경쟁사 가격을 받아 조정 권고를 반환합니다",
      "inputModes": ["text"],
      "outputModes": ["text"]
    }
  ]
}

# 1. 가격 수집 에이전트 (Kafka Producer)
# MCP를 통해 경쟁사 웹사이트 스크래핑 후 이벤트 발행
from confluent_kafka import Producer
import json
import uuid
 
class PriceCollectorAgent:
    def __init__(self):
        self.producer = Producer({'bootstrap.servers': 'kafka:9092'})
        self.mcp_client = MCPClient("scraping-server")
 
    async def collect_and_publish(self, competitor_url: str):
        # MCP 도구 호출: 경쟁사 가격 스크래핑
        price_data = await self.mcp_client.call_tool(
            "scrape_price",
            {"url": competitor_url}
        )
 
        event = {
            "event_type": "competitor_price_changed",
            "competitor": competitor_url,
            "new_price": price_data["price"],
            "product_id": price_data["product_id"],
            "timestamp": price_data["timestamp"],
            "correlation_id": str(uuid.uuid4())  # 분산 추적용, 처음부터 반드시 포함
        }
 
        self.producer.produce(
            topic="ecommerce.price-changed.v1",
            key=price_data["product_id"],
            value=json.dumps(event)
        )
        self.producer.flush()

# 2. 가격 분석 에이전트 — A2A Task 위임
# 주의: 아래는 A2A 프로토콜 흐름을 표현한 의사 코드입니다.
# 실제 구현은 google-a2a 레퍼런스 SDK 또는 httpx를 직접 사용해야 합니다.
import asyncio
import httpx
from confluent_kafka import Consumer
 
class PriceAnalysisAgent:
    def __init__(self):
        self.consumer = Consumer({
            'bootstrap.servers': 'kafka:9092',
            'group.id': 'price-analysis-group',
            'auto.offset.reset': 'earliest'
        })
        self.consumer.subscribe(['ecommerce.price-changed.v1'])
        # MCP로 내부 재고 DB에도 접근
        self.mcp_client = MCPClient("inventory-db-server")
 
    async def process(self):
        loop = asyncio.get_event_loop()
        while True:
            # confluent_kafka poll()은 동기 호출이므로
            # asyncio 이벤트 루프 블로킹 방지를 위해 executor로 래핑
            msg = await loop.run_in_executor(None, self.consumer.poll, 1.0)
            if msg is None or msg.error():
                continue
 
            event = json.loads(msg.value())
 
            # MCP로 내부 재고 상태 확인
            inventory = await self.mcp_client.call_tool(
                "get_inventory", {"product_id": event["product_id"]}
            )
 
            should_adjust = self.analyze_price_gap(event, inventory)
 
            if should_adjust:
                # A2A: Agent Card 조회 후 주문 업데이트 에이전트에 위임
                async with httpx.AsyncClient() as client:
                    await client.post(
                        "http://order-update-agent/tasks/send",
                        json={
                            "message": {
                                "parts": [{
                                    "text": f"Update price for {event['product_id']} to {event['new_price'] * 0.99}"
                                }]
                            },
                            "metadata": {"correlation_id": event["correlation_id"]}
                        }
                    )
 
            # 정상 처리 후 오프셋 커밋
            self.consumer.commit()

// Flink Java: IoT 센서 스트림에서 이상 패턴 감지
// 10분 윈도우 내 진동 센서 평균이 임계값 초과 시 이벤트 발행
DataStream<SensorReading> sensorStream = env
    .addSource(new FlinkKafkaConsumer<>(
        "manufacturing.sensor-reading.v1",
        new SensorReadingDeserializer(),
        kafkaProps
    ));
 
DataStream<MaintenanceAlert> alerts = sensorStream
    .keyBy(SensorReading::getMachineId)
    // 10분 슬라이딩 윈도우, 1분마다 갱신
    .window(SlidingEventTimeWindows.of(
        Time.minutes(10), Time.minutes(1)
    ))
    .aggregate(new VibrationAnomalyDetector())
    .filter(alert -> alert.getSeverity() > THRESHOLD);
 
// 이상 감지 이벤트를 Kafka 토픽에 발행
// 진단 에이전트가 구독하고 A2A로 유지보수 에이전트에 위임
alerts.addSink(new FlinkKafkaProducer<>(
    "agent.task-requests.maintenance",
    new AlertSerializer(),
    kafkaProps
));

IoT 센서
  → Kafka(manufacturing.sensor-reading.v1)
  → Flink(10분 윈도우 이상 패턴 감지)
  → Kafka(agent.task-requests.maintenance)
  → 진단 에이전트
  → A2A
  → 유지보수 스케줄링 에이전트

# 권장 토픽 네이밍 패턴: {도메인}.{이벤트-타입}.{버전}
 
topics:
  # 원시 이벤트 (에이전트가 발행)
  - ecommerce.price-changed.v1
  - ecommerce.order-updated.v1
  - manufacturing.sensor-reading.v1
 
  # 에이전트 간 작업 위임
  - agent.task-requests.pricing
  - agent.task-results.pricing
  - agent.task-requests.maintenance
  - agent.dead-letter.v1          # 반복 실패 메시지 격리용
 
  # 알림 및 액션
  - notification.customer-alert.v1
  - action.price-adjustment.v1
 
# Schema Registry에 Avro 스키마 등록
# 에이전트 메시지 형식 변경 시 하위 호환성 보장
schemas:
  price-changed:
    type: record
    fields:
      - name: product_id
        type: string
      - name: old_price
        type: double
      - name: new_price
        type: double
      - name: source_agent
        type: string
      - name: correlation_id    # 분산 추적용 — 처음부터 반드시 포함
        type: string