medisolveai_kafka/
├── core/           # BaseEvent, KafkaConfig, Exceptions
├── broker/         # AsyncProducer, AsyncConsumer
├── events/         # 이벤트 스키마 (도메인/버전별)
│   └── bay/v1/     # SupplyDispatch, StockReceive, LowStockAlert
├── mcp/            # MCP 서버 (Claude Code 연동)
└── example/        # 예제 앱 (Docker + FastAPI UI)

pip install medisolveai-kafka

KAFKA_BOOTSTRAP_SERVERS=your-kafka:9092   # 필수
KAFKA_GROUP_ID=my-service-group           # Consumer 사용 시 필수

from contextlib import asynccontextmanager
from fastapi import FastAPI
from medisolveai_kafka import AsyncProducer
from medisolveai_kafka.events.bay.v1.dispatch import SupplyDispatch

producer = AsyncProducer()

@asynccontextmanager
async def lifespan(app: FastAPI):
    await producer.start()
    yield
    await producer.stop()

app = FastAPI(lifespan=lifespan)

@app.post("/dispatch")
async def dispatch():
    await producer.send(SupplyDispatch(type="product", id=1, quantity=10))

from medisolveai_kafka import AsyncConsumer
from medisolveai_kafka.events.bay.v1.dispatch import SupplyDispatch

consumer = AsyncConsumer()

@consumer.on(SupplyDispatch)
async def handle(event: SupplyDispatch):
    print(f"소모발송: {event.type} id={event.id} qty={event.quantity}")

# 기본: 처리 성공 후 commit (안전, 유실 없음)
@consumer.on(SupplyDispatch)

# 즉시 커밋: 수신 즉시 commit (처리량 우선, 유실 허용)
@consumer.on(StockReceive, commit_strategy="immediate")

@consumer.on(
    SupplyDispatch,
    max_retries=5,              # 재시도 횟수 (기본값: 3)
    retry_backoff_ms=1000,      # 재시도 간격 (기본값: 300ms, exponential)
    dlq_enabled=False,          # DLQ 비활성화
    commit_strategy="after_process",
)
async def handle(event: SupplyDispatch):
    ...

send() -> serialize() -> Kafka 전송
                          ↓ 실패
                     재시도 (exponential backoff)
                          ↓ 전부 실패
                     DLQ 토픽 전송

메시지 수신 -> 핸들러 실행 -> 성공 -> offset commit
                            -> 실패 -> 재시도
                                        -> 전부 실패 -> DLQ -> commit

메시지 수신 -> offset commit -> 핸들러 실행 (실패해도 재처리 없음)

{
  "mcpServers": {
    "medisolveai-kafka": {
      "command": "uvx",
      "args": ["medisolveai-kafka-mcp"]
    }
  }
}

cd my-service-project
claude
# Claude Code가 medisolveai-kafka MCP 서버를 인식

"bay 소모발송 이벤트 보내는 코드 짜줘"
→ Claude가 list_events → get_event_schema → get_producer_usage 호출
→ 규격에 맞는 코드 생성

"어떤 이벤트 사용할 수 있어?"
→ Claude가 list_events 호출
→ [bay/v1] SupplyDispatch, StockReceive, LowStockAlert 응답

"StockReceive 스키마 알려줘"
→ Claude가 get_event_schema 호출
→ 필드, 타입, 비즈니스 규칙, import 경로 응답

"Consumer 즉시커밋으로 구현해줘"
→ Claude가 get_consumer_usage 호출
→ commit_strategy="immediate" 포함한 코드 생성

담당자 서비스에서 Claude Code 실행
    ↓
MCP 서버가 medisolveai-kafka 라이브러리의 events/ 자동 스캔
    ↓
담당자가 자연어로 요청
    ↓
Claude가 MCP 도구로 이벤트/스키마/사용법 조회
    ↓
규격에 맞는 코드 생성 (import 경로, 필수 필드, 비즈니스 룰 포함)

docker compose -f medisolveai_kafka/example/docker-compose.example.yml up -d
# http://localhost:8000  - 예제 UI (Producer/Consumer 시나리오 테스트)
# http://localhost:8080  - Redpanda Console (토픽/메시지 확인)

# medisolveai_kafka/events/{도메인}/v1/{행위}.py
from typing import ClassVar, Literal
from medisolveai_kafka.core.base import BaseEvent

class NewEvent(BaseEvent):
    TOPIC: ClassVar[str] = "domain.action.v1"
    VERSION: ClassVar[str] = "1.0"

    # 비즈니스 필드
    field: str

    def validate_business_rules(self) -> None:
        if not self.field:
            raise ValueError("field is required")

uv sync --all-extras       # 의존성 설치
uv run pytest tests/ -v    # 테스트 실행 (57 tests)