SYSTEM_DESIGN.md의 추상 설계를 실제 OSS로 구현할 때, 각 컴포넌트가 어떤 프로토콜·토픽·이벤트로 연결되는지를 정리한 문서.

┌───────────┐    RTSP    ┌─────────┐ MQTT (frigate/events)  ┌──────────────┐
│ CCTV (n)  │ ─────────► │ Frigate │ ─────────────────────► │              │
└───────────┘            └─────────┘                        │              │
                                                            │  Mosquitto   │
┌───────────┐  Wi-Fi/SDK  ┌──────────────┐  MQTT             │  (MQTT       │
│ SPOT      │ ◄──────────►│ Robot Adapter│ ◄────────────────►│   broker)    │
└───────────┘            │ (MCP server) │                    │              │
                         └──────────────┘                    │              │
┌───────────┐  WSS/REST   ┌──────────────┐                   │              │
│ Tablet x3 │ ◄──────────►│ Home         │ ◄────────────────►│              │
└───────────┘            │ Assistant    │   MQTT             │              │
                         │ (state/UI)   │                    │              │
                         │ + SQLite     │                    │              │
                         └──────────────┘                    └──────┬───────┘
                                                                    │ MQTT
                                                                    ▼
                                                            ┌───────────────┐
                                                            │ LangGraph     │
                                                            │ Orchestrator  │
                                                            │  ┌──────────┐ │
                                                            │  │ Claude   │ │
                                                            │  │ (tool    │ │
                                                            │  │  use)    │ │
                                                            │  └──────────┘ │
                                                            │  via MCP:     │
                                                            │   • robot.*   │
                                                            │   • tablet.*  │
                                                            │   • ha.*      │
                                                            └───────────────┘

핵심 원칙:

• MQTT broker가 모든 도메인 이벤트의 hub. 모든 컴포넌트는 broker로만 통신.
• HA의 SQLite가 영속 상태. Context Server를 별도로 두지 않고 HA의 entity/state 머신을 그대로 사용.
• LangGraph는 MQTT 구독자 + Claude 호출자. 자신의 상태는 거의 갖지 않음(stateless graph + 외부 상태는 HA에서 조회).

Frigate는 객체 탐지 시 자동으로 MQTT 이벤트를 발행한다.

frigate.yml (요지):

mqtt:
  host: mosquitto
  port: 1883
  topic_prefix: frigate
  client_id: frigate

cameras:
  cafe_cam_01:
    ffmpeg:
      inputs:
        - path: rtsp://camera-cafe-01/stream
          roles: [detect, record]
    detect: { width: 1280, height: 720, fps: 5 }
    objects:
      track: [person]
    zones:
      cafe_zone:
        coordinates: "0,0,1280,0,1280,720,0,720"

발행되는 MQTT 토픽:

• frigate/events — 객체 진입/이탈/업데이트 (JSON payload, before/after 스냅샷 포함)
• frigate/{camera}/person — 카메라별 person 카운트
• frigate/{camera}/{zone} — 존 점유 여부 (ON/OFF)

우리 도메인 토픽으로의 매핑: HA 자동화 또는 작은 bridge 서비스가 frigate/events → vision/zone/enter / vision/zone/exit로 변환.

# bridge.py (의사코드, 30라인 정도)
@mqtt.on("frigate/events")
def on_frigate_event(msg):
    e = json.loads(msg.payload)
    if e["type"] == "new":
        publish("vision/zone/enter", {
            "session_id": resolve_session(e["after"]),  # ReID 매칭
            "zone_id": map_camera_to_zone(e["after"]["camera"]),
            "confidence": e["after"]["score"],
            "ts": e["after"]["frame_time"],
        })

Cross-camera ReID(Re-identification, 보행자 재식별)는 Frigate 기본 기능 부족 → BoxMOT/Norfair 보강 또는 단순 외형 임베딩 매칭으로 직접 구현.

전시장 데모 한정. 브라우저는 RTSP를 직접 재생할 수 없으므로 RTSP를 브라우저 호환 전송으로 바꿔주는 중계서버(relay) 가 필요하다. 이를 독립 서비스 cctv-relay(go2rtc 기반) 로 둔다 — Frigate(detect/이벤트)와 분리해 라이브 영상만 책임진다. 배경·불변식 완화 근거는 SYSTEM_DESIGN.md §8.4, 서비스 상세는 SERVICES.md §2.5.

중계 흐름:

CCTV n ──RTSP(config 구동 자동 연결)──► cctv-relay(go2rtc) ──► WebRTC/MSE restream
CCTV n ──RTSP──► Frigate ──► detect 파이프라인 (frigate/events)     ※ 비전은 별도 경로
브라우저(tablet-fe) ──① WHEP offer(SDP)──► tablet-fe BFF(same-origin) ──► cctv-relay /api/webrtc
브라우저 ◄══② 미디어(SRTP, ICE P2P)══► cctv-relay   (Node BFF는 시그널링만 중계, 미디어 미경유)

• n-stream config 구동: config/cctv-relay.yaml의 streams: 맵에 원본 RTSP 주소를 한 줄 추가하면 자동 등록·연결(코드 변경 불필요).
• 시그널링은 BFF 프록시: 브라우저는 카메라 자격증명·relay 주소를 모른 채 same-origin(/api/cctv/webrtc/<cam>)으로만 WHEP offer를 보낸다 → 자격증명은 서버(cctv-relay/BFF)에만(§3.1의 {ENV_VAR} 치환과 동일 원칙).
• 미디어는 P2P: SDP 협상 후 영상은 브라우저↔cctv-relay ICE로 직접. LAN 내부이므로 STUN/TURN 불필요(host candidate IP:8555).
• Frigate와 분리: 기본은 cctv-relay가 카메라에서 독립 pull → 비전 파이프라인과 장애 격리(단일-pull 변형 트레이드오프는 SERVICES §2.5).
• 코덱: WebRTC는 H.264만 → H.265 카메라는 go2rtc transcode 필요(CPU/GPU 비용).

HA는 MQTT integration을 활성화하면 broker의 토픽을 entity에 자동 매핑할 수 있다.

MQTT Discovery로 엔티티 자동 등록 (LangGraph 또는 Robot Adapter가 발행):

Topic: homeassistant/sensor/spot_01_battery/config
Payload: {
  "name": "SPOT-01 Battery",
  "state_topic": "robot/spot-01/state",
  "value_template": "{{ value_json.battery_pct }}",
  "unit_of_measurement": "%",
  "device": { "identifiers": ["spot-01"], "name": "SPOT 01" }
}

핵심 엔티티 모델:

• sensor.spot_01_battery, sensor.spot_01_pose, binary_sensor.spot_01_online
• sensor.session_a_zone, sensor.session_a_mood
• binary_sensor.cafe_occupied (Frigate zone에서 자동 생성)

상태 영속: HA 기본 recorder가 SQLite에 자동 기록 — 별도 Context Server 불필요.

대시보드: Lovelace YAML로 운영자 콘솔 즉시 구성 가능 (그래프, 카메라 스트림, 로봇 상태 카드).

서비스 호출 (HA → 외부): mqtt.publish 서비스로 임의 토픽 발행 가능 → 운영자가 대시보드 버튼으로 비상 정지·로봇 호출 등 수동 명령.

LangGraph 서비스의 책임은 둘:

1. MQTT 구독: vision/zone/+, tablet/+/intent, context/+/+/update 등에 반응.
2. Claude tool use 호출: 컨텍스트를 모아 Anthropic API 호출, 도구 호출 결과를 MQTT publish 또는 MCP 호출로 실행.

LangGraph 노드 구조:

from langgraph.graph import StateGraph
from anthropic import Anthropic

class FlowState(TypedDict):
    trigger_event: dict
    context_snapshot: dict
    llm_decisions: list[dict]

graph = StateGraph(FlowState)
graph.add_node("assemble_context", assemble_from_ha)   # HA REST API로 현재 상태 조회
graph.add_node("call_claude", call_with_tools)         # Claude Agent SDK 호출
graph.add_node("dispatch", dispatch_tool_calls)        # MCP 또는 MQTT publish
graph.add_node("guardrail", validate_actions)          # 화이트리스트·hard clamp

graph.set_entry_point("assemble_context")
graph.add_edge("assemble_context", "call_claude")
graph.add_edge("call_claude", "guardrail")
graph.add_edge("guardrail", "dispatch")

Claude Agent SDK 호출 (도구는 MCP로 등록):

from anthropic import Anthropic
client = Anthropic()

response = client.messages.create(
    model="claude-opus-4-7",
    max_tokens=1024,
    system=ZONE_PLAYBOOK_SYSTEM_PROMPT,  # prompt cache 적용
    tools=[
        {"type": "mcp", "server": "robot-adapter"},
        {"type": "mcp", "server": "tablet-gateway"},
        {"type": "mcp", "server": "home-assistant"},
    ],
    messages=[{"role": "user", "content": context_message}],
)

도구를 LangGraph 안에 직접 함수로 두지 않고 MCP 서버 3개로 분리하면 재사용성·테스트·권한 분리가 용이.

MCP 서버 구성:

mcp-robot-adapter (Python) — SPOT SDK 래핑

# 노출 도구 (예시)
@mcp.tool
def robot_command(robot_id: str,   # 런타임에 config/robots.yaml 레지스트리로 검증
                  capability: Literal["goto", "follow", "stop", "look_at", "gesture"],
                  params: dict) -> dict:
    """로봇에 capability 명령을 발행. 안전 인터록을 통과해야 실행."""
    if not safety_interlock.allow(robot_id, capability, params):
        return {"status": "rejected", "reason": "safety_interlock"}
    return adapters[robot_id].execute(capability, params)

mcp-tablet-gateway (Node.js) — 태블릿에 옵션·메시지 push

@tool
async function showOptions(tabletId: string, options: Option[]): Promise<Ack> {
  await ws.send(tabletId, { type: "show_options", options });
  return { ok: true };
}

mcp-home-assistant — HA 상태 조회·서비스 호출 (커뮤니티 구현체 활용 가능)

• ha.get_state("sensor.session_a_mood")
• ha.call_service("light.turn_on", { ... })

LangGraph는 위 MCP 서버 3개를 Claude에 도구로 노출해, Claude가 호출하면 실제 액션이 실행된다.

도구로 명령을 받는 것과 별도로, 로봇은 자기 상태를 1Hz로 MQTT에 발행해야 HA·Frigate·LangGraph가 모두 본다.

Topic: robot/spot-01/state
Payload: {
  "robot_id": "spot-01",
  "pose": { "x": 2.4, "y": 1.1, "yaw": 0.78, "zone_id": "cafe" },
  "battery_pct": 87,
  "current_action": { "type": "look_at", "started_at": "2026-05-04T10:00:00Z" },
  "faults": [],
  "connection": "online",
  "updated_at": "2026-05-04T10:00:01Z"
}

이 토픽은 HA MQTT Discovery로 자동 entity화되어 Lovelace에 그대로 노출된다.

두 가지 통합 옵션:

옵션 A — HA 네이티브: 태블릿이 HA 대시보드를 풀스크린으로 표시. 빠르지만 커스텀 인터랙션 한계.

옵션 B — 자체 앱: React Native 앱이 HA WebSocket API에 인증 토큰으로 접속, 엔티티 변경을 구독하고 사용자 입력은 MQTT로 publish.

// 태블릿 앱 (옵션 B)
import { connectToHA } from "ha-websocket";
const ha = await connectToHA(URL, TOKEN);
ha.subscribe("state_changed", (e) => {
  if (e.entity_id === `sensor.session_${sessionId}_options`) {
    setOptions(JSON.parse(e.new_state.state));
  }
});

// 사용자가 옵션 선택 시
mqtt.publish(`tablet/${tabletId}/intent`, { option_id: "drink_americano" });

권장: 옵션 B — 인터랙션 자유도가 높고, HA는 백엔드 상태 저장만 담당.

로봇 텔레메트리(특히 SPOT의 카메라/IMU(Inertial Measurement Unit, 관성 측정 장치)/관절 상태)를 시각화할 때 사용. ROS 2를 도입하지 않으면 가치 제한적이므로 운영 단계 진입 후 필요 시 추가.

연결 옵션: foxglove-bridge 또는 MQTT-to-foxglove 어댑터.

1. CCTV(cafe_cam_01) → Frigate
2. Frigate publish → mqtt:frigate/events
   { type:"new", after:{ camera:"cafe_cam_01", label:"person", id:"track_42" } }
3. bridge.py 변환 → mqtt:vision/zone/enter
   { session_id:"sess_A", zone_id:"cafe", confidence:0.97 }
4. HA 자동화 트리거 → sensor.session_a_zone 상태 갱신
   동시에 LangGraph 서비스가 vision/zone/enter 구독 콜백 실행
5. LangGraph: assemble_context
   - HA REST: GET /api/states/sensor.session_a_mood → "neutral"
   - HA REST: GET /api/states/sensor.weather_condition → "sunny"
   - 로봇 상태: GET /api/states/sensor.spot_01_pose
6. LangGraph: Claude tool use 호출 (mcp-robot-adapter, mcp-tablet-gateway 노출)
7. Claude 응답 → tool_calls:
   - robot_command(robot_id="spot-01", capability="look_at", params={target:"sess_A"})
   - showOptions(tablet_id="t01", options=["커피 추천","음악 추천"])
8. guardrail 통과 → MCP 호출 → 실제 실행
9. SPOT 동작, 태블릿 UI 갱신, robot/spot-01/state 1Hz publish
10. HA 대시보드에 모든 것이 라이브로 표시

1. 사용자 "공원에서 산책 같이 하기" 선택
2. 태블릿 → mqtt:tablet/t01/intent { option:"walk_with_me" }
3. LangGraph 구독 콜백 → assemble_context (현 zone=park)
4. Claude → robot_command(spot-01, capability="follow", params={target:"sess_A", max_distance:2.0})
5. mcp-robot-adapter: safety_interlock 검증 (proximity, zone polygon) → SPOT SDK 호출
6. 1Hz robot/spot-01/state 발행 → 태블릿 진행도 표시

services:
  mosquitto:
    image: eclipse-mosquitto:2.0
    ports: ["1883:1883"]
    volumes: ["./mosquitto:/mosquitto/config"]

  frigate:
    image: ghcr.io/blakeblackshear/frigate:stable
    devices: ["/dev/dri:/dev/dri"]   # 또는 NVIDIA GPU
    volumes:
      - ./frigate.yml:/config/config.yml
      - frigate_media:/media/frigate
    ports: ["5000:5000"]

  home-assistant:
    image: ghcr.io/home-assistant/home-assistant:stable
    volumes: ["./ha-config:/config"]
    network_mode: host

  bridge:
    build: ./services/bridge        # frigate.events → vision.* 변환
    depends_on: [mosquitto]

  mcp-robot-adapter:
    build: ./services/mcp-robot-adapter
    devices: ["/dev/spot:/dev/spot"]    # 필요 시
    environment:
      - SPOT_HOSTNAME=...
      - SPOT_USERNAME=...

  mcp-tablet-gateway:
    build: ./services/mcp-tablet-gateway

  langgraph-orchestrator:
    build: ./services/orchestrator
    extra_hosts:
      - "host.docker.internal:host-gateway"     # HA(host network) 도달
    environment:
      - ANTHROPIC_API_KEY=...
      - HA_URL=http://host.docker.internal:8123
      - HA_TOKEN=...
    depends_on: [mosquitto, mcp-robot-adapter, mcp-tablet-gateway]

OSS 백본을 채택하면 자체 구현 필요 영역은 다음 4개로 축소된다.

자체 인프라(MQTT, 상태 저장, 대시보드, NVR(Network Video Recorder, 네트워크 비디오 레코더), 객체 탐지)는 모두 OSS가 제공.

• MQTT QoS(Quality of Service, 서비스 품질 등급): 명령(robot/.../command)은 QoS 1 또는 2, 텔레메트리(robot/.../state)는 0 또는 1.
• HA recorder retention: 기본 10일이지만 이 시스템은 통계 보존이 목적 아니므로 2~3일로 축소 (recorder.purge_keep_days: 3).
• MCP 권한: mcp-robot-adapter는 mTLS(mutual TLS, 상호 인증 TLS) + 토큰. LangGraph만 호출 가능하게 ACL(Access Control List, 접근 제어 목록).
• Frigate ReID 한계: 단일 카메라 추적은 강하지만 cross-camera는 보강 필요 — services/bridge에서 임베딩 매칭 보강 또는 BoxMOT 사이드카.
• HA 자동화 vs LangGraph 책임 경계: 단순 룰(예: 로봇 배터리 < 20% 시 알림)은 HA 자동화로, 컨텍스트 의존 의사결정은 LangGraph로. 같은 결정을 두 곳에서 하면 충돌난다.
• prompt cache: zone playbook + system prompt는 stable → Anthropic prompt cache로 비용·지연 절감.

• Frigate native (frigate/{cam}/{zone} ON/OFF): 단순 zone 점유 표시용. HA가 binary_sensor.<zone>_occupied로 자동 흡수. 운영자 대시보드 상단 "현재 점유 zone" 카드용으로만 사용.
• bridge vision/zone/enter|exit: session_id 매칭된 도메인 이벤트. orchestrator 트리거·세션 zone history 추적용.
• 두 경로가 정보가 겹치는 듯하지만 소비자가 다르므로 의도된 중복. orchestrator가 Frigate native 토픽을 직접 구독하지 않게 ACL로 차단.

• 서비스 이름은 orchestrator (디렉터리 services/orchestrator).
• 내부에서 LangGraph를 사용하지만 "LangGraph 서비스"라 부르지 않음 — LangGraph는 라이브러리·구현 도구.
• 문서·로그·메트릭에서 일관되게 orchestrator로 표기.

• MCP 표준 transport 3종: stdio, SSE, HTTP.
• 본 시스템은 SSE over HTTP 채택:
  • 컨테이너 분리되어 있어 stdio 부적합.
  • SSE는 양방향 long-lived 스트림으로 도구 호출 + 진행 알림에 적합.
  • HA REST와 동일 라우팅(역프록시 단순화).
• mcp-robot-adapter·mcp-tablet-gateway 둘 다 :8000/mcp 엔드포인트 노출. orchestrator가 클라이언트로 연결.
• 인증: opaque bearer token (각 서비스의 .env.secret에 발급), 상호 mTLS는 미적용(LAN 내부).

• enter hold: 800ms — NFR 1.5s 이내 보장.
• exit grace: 3000ms — 사용자가 잠깐 가려져도 exit 안 발동.
• dedup window: 동일 (session_id, zone_id) enter는 1000ms 내 재발행 차단.
• 임계값은 config/bridge.yaml에서 운영 중 조정 가능 (재기동 필요).

• 만료 처리자: mcp-tablet-gateway가 담당.
• ttl_s 만료 시 게이트웨이가 태블릿에 {type: "dismiss_options", message_id} 자동 push → 태블릿 UI가 옵션 영역 비움.
• 만료 후 사용자 입력은 게이트웨이에서 "stale_intent" 응답으로 거부 + Loki 로그 한 줄.
• 만료 전 새 show_options(같은 tablet_id)가 도착하면 이전 message_id는 즉시 폐기.

SYSTEM_DESIGN.md 컴포넌트 → OSS 매핑: