-
Notifications
You must be signed in to change notification settings - Fork 0
ROBOT_ADAPTER_ONBOARDING
mcp-robot-adapter개발에 착수하는 개발자가 MQTT 기초 개념 → broker(Mosquitto) 사용자 등록 → Home Assistant 연동 → 메시지 발행/구독까지 한 번에 따라갈 수 있도록 기존 설계 문서의 관련 내용을 모은 실무 가이드다. MQTT를 처음 접하는 사람을 기준으로 쓰였다 — 이미 익숙하다면 §1은 건너뛰어도 된다. 본 문서는 요약/안내용이며, 충돌 시 원본 설계 문서가 우선한다 — 토픽·페이로드는 MQTT_SCHEMA.md, 서비스 책임·ACL은 SERVICES.md, 시크릿 절차는 DEPLOYMENT.md.
orchestrator ──(MCP tool call, 주 경로)──▶ mcp-robot-adapter ──(bosdyn-client)──▶ SPOT
home-assistant ─(MQTT robot/+/command, 폴백 경로)─▶ mcp-robot-adapter
mcp-robot-adapter ─(MQTT robot/+/state·ack·fault)─▶ mosquitto ─▶ HA·orchestrator 구독
-
adapter ↔ HA 직접 연결은 없다. HA는 MQTT integration으로
robot/+/state등을 구독해sensor.spot_01_battery같은 엔티티로 흡수한다 (엔티티가 무엇인지는 §3.0에서 설명). adapter가 HA REST API를 호출하거나HA_TOKEN을 가질 필요가 없다. - 명령 주 경로는 MCP tool call(MQTT 미경유), MQTT
robot/{robot_id}/command는 HA 폴백 경로 전용이다. 두 entry point는 adapter 내부에서 동일한 priority queue + safety interlock으로 수렴한다. 상세: MQTT_SCHEMA.md §3.4.
⚠️ adapter 개발 중에도 orchestrator가robot/+/command를 발행하도록 만들지 않는다 — ACL에서 orchestrator에게 이 토픽 write 권한 자체가 없다.
MQTT(Message Queuing Telemetry Transport)는 가볍고 단순한 메시지 전달 프로토콜이다. HTTP처럼 "요청하고 응답받는" 방식이 아니라, 우편함 시스템에 가깝다:
발행자(publisher) ──메시지──▶ [ broker (우체국) ] ──메시지──▶ 구독자(subscriber)
mosquitto
| 용어 | 뜻 |
|---|---|
| broker | 모든 메시지가 거쳐 가는 중앙 중계 서버. 이 프로젝트에서는 mosquitto라는 오픈소스 broker를 쓴다. 발행자와 구독자는 서로를 모르고, 오직 broker만 안다. |
| client | broker에 접속하는 모든 프로그램. adapter도, HA도, orchestrator도 전부 client다. 각 client는 client_id(접속 식별 이름)와 username/password를 가진다. |
| publish (발행) | client가 broker에게 "이 메시지를 이 topic으로 보내줘"라고 건네는 행위. 받는 쪽 주소를 몰라도 된다 — topic 이름만 알면 된다. |
| subscribe (구독) | client가 broker에게 "이 topic으로 오는 메시지를 나한테 전달해줘"라고 등록하는 행위. 이후 해당 topic에 발행되는 메시지가 자동으로 도착한다. |
| topic | 메시지의 주소/채널 이름. robot/spot-01/state처럼 /로 계층을 나눈 문자열이다. |
| payload | 메시지의 본문. MQTT 입장에선 그냥 바이트 덩어리이고, 이 프로젝트에서는 항상 JSON 문자열을 넣는다. |
이 구조의 장점: 발행자는 "누가 받는지" 신경 쓰지 않는다. adapter가 robot/spot-01/state에 상태를 발행하면, HA·orchestrator·디버깅 터미널 등 구독 중인 모두가 동시에 받는다. 구독자가 늘어나도 발행 코드는 한 줄도 바뀌지 않는다.
처음 접하면 "토픽을 어디에 등록하지?"부터 찾게 되는데, MQTT 프로토콜 자체에는 topic을 미리 등록·생성하는 절차가 없다. topic은 누군가 그 이름으로 publish 하는 순간 존재하고, 아무도 안 쓰면 그냥 없는 것이다. 데이터베이스 테이블처럼 미리 만들어 두는 개념이 아니다.
다만 이 프로젝트에서 "토픽 등록"이라 부르는 것은 다음 두 가지를 뜻한다:
- 문서 등록 — MQTT_SCHEMA.md의 토픽 카탈로그에 토픽명·payload 형식·QoS·retain을 정의한다. 모든 서비스가 이 문서를 보고 맞춘다 (단일 소스).
- 권한 등록 — broker의 ACL(Access Control List, 접근 제어 목록) 파일에 "어느 사용자가 어느 topic에 읽기/쓰기 가능한지"를 추가한다. 이 프로젝트의 mosquitto는 ACL에 없는 publish를 에러도 없이 조용히 버린다(silent drop). 그래서 "분명 보냈는데 안 보인다"의 1순위 원인이 ACL 누락이다.
구체적 절차는 §5에서 다룬다.
topic 와일드카드 — 구독할 때만 쓸 수 있는 특수 문자 두 가지:
| 문자 | 의미 | 예 |
|---|---|---|
+ |
한 단계를 아무 값이나 매칭 |
robot/+/state → robot/spot-01/state, robot/kaist-01/state 모두 수신 |
# |
그 이하 전부 매칭 (맨 끝에만) |
robot/# → robot으로 시작하는 모든 topic 수신 |
발행할 때는 와일드카드를 쓸 수 없다 — 항상 구체적인 topic 이름으로 발행한다.
publish 할 때마다 topic·payload 외에 다음 옵션을 함께 지정한다. 어떤 값을 쓸지는 토픽마다 MQTT_SCHEMA.md §2에 이미 정해져 있다 — 개발자가 임의로 고르지 않는다.
| QoS | 보장 | 비유 | 이 프로젝트의 용도 |
|---|---|---|---|
| 0 | 최대 1회 — 보내고 끝, 유실 가능 | 일반 우편 |
robot/+/state (1초마다 다시 보내니 한 번 빠져도 무방) |
| 1 | 최소 1회 — 도착 확인, 중복 가능 | 등기 우편 |
robot/+/command·ack·fault (놓치면 안 되는 것) |
| 2 | 정확히 1회 — 가장 무겁다 | 내용증명 |
safety/estop 단 하나 (절대 유실·중복 금지) |
QoS 1은 "중복 도착 가능"이므로 구독자는 envelope의 id(§4.4)로 같은 메시지를 두 번 처리하지 않게 한다 — 이것을 멱등 처리(idempotent) 라 한다.
보통 메시지는 그 순간의 구독자에게만 전달되고 사라진다. retain=true로 발행하면 broker가 그 topic의 마지막 메시지 1개를 보관해 두었다가, 나중에 구독을 시작하는 client에게 즉시 건네준다.
- 예: HA가 재시작돼도
robot/spot-01/state를 구독하는 순간 마지막 로봇 상태를 바로 받는다 — 다음 1Hz 발행을 기다릴 필요가 없다. - 부작용: 발행자가 죽어도 마지막 메시지가 영원히 남는다(stale). 그래서 ↓ LWT가 필요하다.
client가 broker에 접속할 때 미리 맡겨두는 메시지. client가 비정상 종료(네트워크 단절, 크래시)되면 broker가 대신 이 메시지를 발행해 준다.
- adapter는
connection: "offline"이 담긴 state 메시지를 유언장으로 맡긴다 → adapter가 죽으면 broker가 자동으로 "로봇 오프라인"을 발행 → HA 대시보드에 즉시 반영되고, retain 된 stale "online" 상태가 남는 문제도 해결된다. - connect 전에 등록해야 한다 (프로토콜상 CONNECT 패킷에 실려 가기 때문).
Mosquitto는 allow_anonymous false(익명 접속 금지) + password_file + acl_file 구성이다 (SERVICES.md §1). adapter는 robot-adapter 사용자로 접속한다.
# 최초 1회 — passwd 파일 생성과 함께 (기존 파일 있으면 -c 빼고 실행)
mosquitto_passwd -c ./mosquitto/config/passwd robot-adapter # 대화형으로 비밀번호 입력
# 자동화 스크립트에서 쓸 때
mosquitto_passwd -b ./mosquitto/config/passwd robot-adapter '<password>'
# 파일 권한 — mosquitto 컨테이너 uid는 1883
chown 1883:1883 ./mosquitto/config/passwd && chmod 0600 ./mosquitto/config/passwd
# 변경 반영 (재시작 없이 reload)
docker kill -s HUP mosquitto- 비밀번호 원문은
deploy/.env.secret의MOSQUITTO_PWD_ROBOT_ADAPTER에 기입한다 (gitignore,chmod 600). adapter 컨테이너는env_file로 이 값을 받는다. - 회전 주기는 분기 1회 — 절차는 DEPLOYMENT.md §9.2 참고.
mosquitto/config/acl에 아래 블록이 이미 정의되어 있다 (SERVICES.md §1 ACL 예시). 새 토픽이 필요하면 MQTT_SCHEMA.md에 먼저 추가한 뒤 ACL을 갱신한다.
user robot-adapter
topic write robot/+/state
topic write robot/+/ack
topic write robot/+/fault
topic read robot/+/command # HA 폴백 경로 수신
topic read safety/estop
topic read safety/proximity
topic write safety/estop # 자체 감지 시 발행
topic read vision/position/update # proximity guard
topic write system/heartbeat/robot-adapter
원칙은 최소 권한 — adapter는 robot 도메인 + safety + 자기 heartbeat만 만진다. 권한 밖 publish는 broker가 조용히 drop하므로, 개발 중 "메시지가 안 보인다" 증상의 1순위 점검 항목이 ACL이다.
| 항목 | 값 |
|---|---|
| host / port |
mosquitto : 1883 (같은 compose network edge) |
| username | robot-adapter |
| password | env MOSQUITTO_PWD_ROBOT_ADAPTER
|
| TLS | 단일 사이트 LAN 내부에선 미사용 (외부 노출 시 8883 + cert) |
| client_id |
mcp-robot-adapter 고정 권장 |
adapter 개발자가 HA에서 해야 할 일은 두 가지뿐이다. 그 전에, 이 절 전체에서 계속 나오는 "엔티티"가 무엇인지부터 짚고 간다.
Home Assistant는 원래 스마트홈 허브다 — 집 안의 온갖 기기(전구·온도계·스위치·카메라)를 한 화면에 모아 보고, "해 지면 전구 켜기" 같은 자동화를 돌리는 오픈소스. HA는 이 세상의 모든 것을 엔티티(entity) 라는 단위로 표현한다.
엔티티 = "이름이 붙은 상태값 하나" 다.
| 실물 | 엔티티 ID | state (현재 값) |
|---|---|---|
| 거실 전구 | light.living_room |
on |
| 거실 온도계 | sensor.living_room_temp |
23.5 |
| 우리 로봇의 배터리 | sensor.spot_01_battery |
87 |
| 우리 로봇의 연결 여부 | binary_sensor.spot_01_online |
on |
- ID는
<domain>.<object_id>구조다. domain이 값의 종류를 정한다 —sensor는 임의 값,binary_sensor는 on/off 두 가지만. - 엔티티는 state 외에 attributes(부가정보: 단위, friendly_name 등)와 last_updated(마지막 갱신 시각)를 함께 가진다 — §7.2의 REST 응답이 정확히 이 구조다.
왜 필요한가 — MQTT 메시지와의 결정적 차이:
| MQTT 메시지 | HA 엔티티 | |
|---|---|---|
| 성격 | 흘러가는 이벤트 — "방금 이런 일이 있었다" | 보관되는 상태 — "지금 값이 이것이다" |
| 읽는 방법 | 발행되는 순간 구독 중이어야 받는다 | 언제든 현재 값을 조회할 수 있다 |
| 과거 기록 | 남지 않는다 (retain 1개 제외) | recorder가 SQLite에 히스토리 자동 기록 |
adapter가 1Hz로 발행하는 robot/spot-01/state는 그 자체로는 흘러가 버리는 스트림이다. HA가 이것을 엔티티로 흡수하는 순간, 추가 코드 없이 다음을 전부 얻는다:
-
언제든 조회 — orchestrator가 LLM 결정 직전에
GET /api/states/sensor.spot_01_battery로 최신 값을 읽는다 (§7.2) - 히스토리 — recorder가 SQLite에 자동 기록 → 배터리 그래프가 공짜
- 대시보드 — Lovelace 카드에 올려 운영자가 실시간으로 본다
-
자동화 트리거/조건 — 폴백 자동화가
binary_sensor.orchestrator_alive같은 엔티티를 조건으로 동작한다 (SERVICES.md §3)
이것이 설계 invariant "Home Assistant가 상태 저장소다" 의 실체다 — 별도의 context server·Redis·DB를 만들지 않고, 엔티티 + recorder가 시스템의 영속 계층 역할을 한다. adapter 개발자 입장에서 한 문장으로 줄이면: state 메시지를 규격대로 잘 발행하기만 하면, 저장·조회·시각화·자동화는 HA가 다 해준다.
그래서 §3.2의 "엔티티 등록"이 필요하다 — HA에게 "robot/spot-01/state topic에서 battery_pct 필드를 뽑아 sensor.spot_01_battery라는 엔티티로 관리해라"라고 알려주는 작업이다.
HA 자체도 Mosquitto의 한 클라이언트다. homeassistant 사용자를 §2.1과 같은 방법으로 passwd에 등록하고, ha-config/configuration.yaml에서 참조한다:
mqtt:
broker: mosquitto
username: !secret mqtt_user # = homeassistant
password: !secret mqtt_password
discovery: true
discovery_prefix: homeassistanthomeassistant 사용자의 ACL(read robot/+/state·robot/+/ack·robot/+/fault, write robot/+/command 등)도 SERVICES.md §1에 이미 정의되어 있다.
adapter가 기동 시 discovery config를 발행하면 HA가 엔티티를 자동 생성한다 (OSS_INTEGRATION.md §3.2). 아래에 동작 원리부터 예제 코드까지 차례로 설명한다.
원래 HA에 센서를 추가하려면 HA 쪽 YAML 파일을 고치고 HA를 재시작해야 한다. MQTT Discovery는 이 설정 자체를 MQTT 메시지로 보내는 방식이다:
- HA는 MQTT integration을 켜는 순간(§3.1의
discovery: true)homeassistant/+/+/config패턴을 항상 구독하고 있다. - 어떤 client든 이 패턴에 맞는 topic으로 "엔티티 설정 JSON"을 발행하면,
- HA가 그 JSON을 읽고 엔티티를 즉시 생성한다 — HA 재시작도, YAML 수정도 필요 없다.
즉 "로봇이 늘어나면 HA 설정을 고친다"가 아니라, **"adapter가 자기 로봇의 엔티티를 스스로 신고한다"**가 된다. config/robots.yaml에 로봇이 추가돼도 HA 쪽 작업이 0이 되는 것이 이 방식을 쓰는 이유다.
homeassistant / <component> / <object_id> / config
└─ §3.1의 └─ 엔티티 종류 └─ 엔티티 고유 └─ 고정 접미사
discovery_prefix (sensor, 이름
binary_sensor, ...)
| 구성 요소 | 설명 | 예 |
|---|---|---|
discovery_prefix |
§3.1 설정값. 기본 homeassistant
|
homeassistant |
component |
HA 엔티티 도메인 — sensor(숫자/문자 값), binary_sensor(ON/OFF), button, switch 등 |
sensor |
object_id |
이 엔티티의 고유 이름 (HA 내부 식별용) | spot_01_battery |
발행 시 retain=true가 사실상 필수다 — HA가 재시작하면 구독을 다시 시작하는데, retain 되어 있어야 그 순간 broker가 config를 다시 건네주어 엔티티가 복원된다(§1.3 retain 참고). retain이 없으면 HA 재시작 때마다 엔티티가 사라진다.
엔티티 삭제도 같은 topic으로 한다 — 빈(null) payload를 retain으로 발행하면 HA가 해당 엔티티를 제거한다 (§6.3의 retain 삭제와 같은 규약).
{
"name": "SPOT-01 Battery",
"unique_id": "zerone_spot_01_battery",
"state_topic": "robot/spot-01/state",
"value_template": "{{ value_json.data.battery_pct }}",
"unit_of_measurement": "%",
"device": { "identifiers": ["spot-01"], "name": "SPOT 01", "manufacturer": "Boston Dynamics" }
}| 필드 | 설명 |
|---|---|
name |
HA UI에 표시될 이름 |
unique_id |
HA 엔티티 레지스트리 등록용 고유 ID. 이걸 넣어야 운영자가 HA UI에서 이름 변경·비활성화 등을 관리할 수 있다 |
state_topic |
HA가 값을 읽어올 topic. discovery는 "어디서 읽을지"만 알려주고, 실제 값은 여전히 robot/{robot_id}/state 발행(§4 STEP 7)이 공급한다 |
value_template |
state_topic payload(JSON)에서 값을 뽑아내는 Jinja2 템플릿. value_json이 파싱된 payload 전체 — envelope 때문에 value_json.data.…로 접근해야 한다 (§7.3 함정 참고) |
device |
같은 identifiers를 가진 엔티티들을 HA UI에서 하나의 기기로 묶어 보여준다 (SPOT 01 기기 아래 battery·pose·online이 모임) |
- 기동 시(connect 직후) 1회: retain=true로 발행해 두면 평상시엔 이걸로 충분하다.
-
HA 재시작 감지(권장): HA는 자신이 기동되면
homeassistant/statustopic에online을 발행한다(birth message). adapter가 이 topic을 구독해 두었다가online수신 시 discovery config를 재발행하면, retain이 어떤 이유로 유실돼도(broker 볼륨 초기화 등) 엔티티가 자동 복구된다.
# 예시 — services/mcp-robot-adapter/src/adapter/ha_discovery.py (발췌 수준)
import json
DISCOVERY_PREFIX = "homeassistant"
def _device_block(robot_id: str) -> dict:
# 같은 identifiers → HA UI에서 엔티티들이 한 기기로 묶임
return {"identifiers": [robot_id], "name": robot_id.upper(), "manufacturer": "Boston Dynamics"}
def _entity_configs(robot_id: str) -> dict[str, dict]:
"""{config_topic: payload} — 이 로봇이 신고할 엔티티 목록.
state_topic은 전부 robot/{robot_id}/state 하나다 — 1Hz 텔레메트리(§4 STEP 7)
한 메시지에서 value_template으로 각자 다른 필드를 뽑아 쓴다.
"""
oid = robot_id.replace("-", "_") # spot-01 → spot_01 (object_id 규칙: 영숫자/_)
state_topic = f"robot/{robot_id}/state"
device = _device_block(robot_id)
return {
f"{DISCOVERY_PREFIX}/sensor/{oid}_battery/config": {
"name": f"{robot_id.upper()} Battery",
"unique_id": f"zerone_{oid}_battery",
"state_topic": state_topic,
"value_template": "{{ value_json.data.battery_pct }}",
"unit_of_measurement": "%",
"device": device,
},
f"{DISCOVERY_PREFIX}/sensor/{oid}_pose/config": {
"name": f"{robot_id.upper()} Pose",
"unique_id": f"zerone_{oid}_pose",
"state_topic": state_topic,
# 문자열로 합쳐 표시 — 좌표는 world meters (SYSTEM_DESIGN.md §8.3)
"value_template": (
"{{ value_json.data.pose.x | round(2) }}, "
"{{ value_json.data.pose.y | round(2) }}"
),
"device": device,
},
f"{DISCOVERY_PREFIX}/binary_sensor/{oid}_online/config": {
"name": f"{robot_id.upper()} Online",
"unique_id": f"zerone_{oid}_online",
"state_topic": state_topic,
# LWT(§4 STEP 3)가 connection: offline을 발행하므로 이 필드만 보면 된다
"value_template": "{{ 'ON' if value_json.data.connection == 'online' else 'OFF' }}",
"device_class": "connectivity",
"device": device,
},
}
def publish_discovery(client, robot_id: str) -> None:
"""기동 시 1회 + HA birth message 수신 시 호출."""
for topic, payload in _entity_configs(robot_id).items():
client.publish(topic, json.dumps(payload), qos=1, retain=True)
def remove_discovery(client, robot_id: str) -> None:
"""로봇 철수 시 — 빈 payload retain 발행으로 HA 엔티티 삭제."""
for topic in _entity_configs(robot_id):
client.publish(topic, payload=None, qos=1, retain=True)§4의 connect 흐름에 끼워 넣는 위치:
# main.py — STEP 4(connect)와 STEP 5(subscribe) 사이/직후
client.connect("mosquitto", 1883)
client.loop_start()
for robot_id in robot_registry: # config/robots.yaml 레지스트리
publish_discovery(client, robot_id) # 기동 시 1회
client.subscribe("homeassistant/status", qos=1) # HA 재시작 감지
def on_message(client, userdata, msg):
if msg.topic == "homeassistant/status" and msg.payload == b"online":
for robot_id in robot_registry:
publish_discovery(client, robot_id) # HA가 재기동했으니 재신고
return
... # robot/+/command 등 기존 처리
⚠️ discovery config payload는 CloudEvents envelope 예외다.homeassistant/.../config의 payload 형식은 HA가 정의한 것이므로 envelope으로 감싸면 HA가 해석하지 못한다. envelope 규칙(§4 STEP 6)은 우리 도메인 토픽(robot/,safety/등)에만 적용된다.
⚠️ ACL 추가 필요. SERVICES.md §1의 현재 ACL에는 discovery 토픽 권한이 없다 — discovery를 실제로 쓰려면robot-adapter에topic write homeassistant/+/+/config·topic read homeassistant/status,homeassistant사용자에topic read homeassistant/#추가가 필요하다(§5.1 절차대로 설계 문서부터 갱신). ACL 변경 없이 가려면 아래 수동 정의 방식을 쓴다.
결과 엔티티: sensor.spot_01_battery, sensor.spot_01_pose, binary_sensor.spot_01_online. discovery 대신 ha-config/packages/robots.yaml에 수동 정의해도 된다(내용은 config payload와 동일한 필드를 YAML로 적는 것) — 로봇 대수가 고정이면 이쪽이 더 단순하다. 어느 쪽이든 상태의 단일 소스는 robot/{robot_id}/state 토픽이다.
참고: HA long-lived access token(
HA_TOKEN)은 orchestrator 전용이다 (HA REST/WS 호출용). adapter에는 불필요하므로 발급받지 않는다. 발급 절차가 필요하면 DEPLOYMENT.md §9.2.
처음부터 끝까지의 흐름은 8단계다. 각 STEP을 아래에서 하나씩 설명한다.
[STEP 1] 접속 정보 준비 (host·계정) ← §2에서 만든 것
[STEP 2] client 객체 생성 + 인증 정보 설정
[STEP 3] LWT(유언장) 등록 ← 반드시 connect "전"
[STEP 4] connect — broker에 접속
[STEP 5] subscribe — 받을 topic 등록
[STEP 6] payload 만들기 — CloudEvents envelope로 감싼 JSON
[STEP 7] publish — topic·payload·QoS·retain 지정해 발행
[STEP 8] 확인 — 터미널 구독으로 실제 도착 검증
§2에서 만든 계정과 §2.3의 연결 정보를 환경변수로 받는다. 코드에 비밀번호를 하드코딩하지 않는다.
Python에서는 paho-mqtt 라이브러리를 쓴다 (이 프로젝트의 표준 의존성). client 객체를 만들고 client_id와 username/password를 설정한다.
import paho.mqtt.client as mqtt
import os
client = mqtt.Client(client_id="mcp-robot-adapter") # broker가 이 이름으로 나를 식별
client.username_pw_set("robot-adapter", os.environ["MOSQUITTO_PWD_ROBOT_ADAPTER"])-
client_id는 broker 입장에서의 내 이름이다. 같은 client_id로 두 프로세스가 접속하면 먼저 붙은 쪽이 끊긴다 — 디버깅 중 로컬에서 하나 더 띄울 때 주의.
robot/+/state는 retain=true이므로(§1.3), adapter가 비정상 종료하면 마지막 "online" 상태가 broker에 영원히 남는다. 이를 막기 위해 connect 전에 connection: "offline" payload를 유언장으로 등록한다 (MQTT_SCHEMA.md §4.5).
import json
from topics.envelope import wrap # packages/topics — envelope 생성 공용 함수 (STEP 6 참고)
will = wrap("robot.state",
{"robot_id": "spot-01", "connection": "offline", "timestamp": now_rfc3339()},
source="/services/mcp-robot-adapter", subject="robot/spot-01")
client.will_set( # "내가 죽으면 이걸 대신 발행해줘"
topic="robot/spot-01/state",
payload=json.dumps(will),
qos=1,
retain=True, # 죽은 뒤에도 "offline"이 retain 되어야 하므로 true
)client.connect("mosquitto", 1883) # host, port
client.loop_start() # 백그라운드 네트워크 스레드 시작-
connect()시점에 broker가passwd파일로 계정을 검증한다. 계정이 틀리면 여기서 연결 거부된다. -
loop_start()는 송수신을 처리하는 백그라운드 스레드를 띄운다 — 이걸 빼먹으면 publish가 전송되지 않고 subscribe 콜백도 호출되지 않는다 (초심자가 가장 자주 빠지는 함정).
adapter가 받아야 하는 메시지를 구독한다. 와일드카드 +(§1.2)로 모든 로봇 ID를 한 번에 커버한다.
client.subscribe("robot/+/command", qos=1) # HA 폴백 경로 명령
client.subscribe("safety/estop", qos=2) # 절대 누락 금지
client.subscribe("safety/proximity", qos=1)
client.subscribe("vision/position/update", qos=0) # proximity guard 입력 (5Hz)
def on_message(client, userdata, msg):
event = json.loads(msg.payload) # envelope 파싱
... # 같은 priority queue + safety interlock으로 투입
client.on_message = on_message수신한 robot.command는 MCP 경로 명령과 같은 priority queue(safety > operator > llm) + safety interlock을 통과시킨다. expires_at이 지난 채 큐에 남은 명령은 폐기한다.
자체 발행하는 모든 메시지는 CloudEvents 1.0 envelope으로 감싼다 (MQTT_SCHEMA.md §1). envelope이란 실제 내용(data)을 표준화된 메타데이터로 감싼 봉투다 — 누가(source)·언제(time)·무슨 종류(type)의 메시지인지 payload를 열어보지 않고도 알 수 있게 한다.
topic은 / 구분, envelope type은 . 구분이다 (topic robot/spot-01/state ↔ type robot.state).
{
"specversion": "1.0",
"type": "robot.state",
"source": "/services/mcp-robot-adapter",
"id": "01HW9XJ1A2B3C4D5E6F7G8H9JK",
"time": "2026-05-04T10:00:00.123Z",
"datacontenttype": "application/json",
"subject": "robot/spot-01",
"traceparent": "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01",
"data": { "robot_id": "spot-01", "battery_pct": 87, "connection": "online", "...": "..." }
}필드별 의미:
| 필드 | 설명 |
|---|---|
type |
메시지 종류 (dot-notation). 토픽 카탈로그에 정의된 값 |
source |
발행한 서비스. adapter는 항상 /services/mcp-robot-adapter
|
id |
ULID — 시간순 정렬 가능한 고유 ID. 구독자가 중복 수신을 걸러내는 멱등 처리 키 |
time |
now()가 아닌 이벤트 원천 시각 (예: 텔레메트리를 캡처한 순간) |
traceparent |
OpenTelemetry 분산 트레이싱 연결 고리 — 가능하면 항상 포함 |
data |
실제 내용. 필드 정의는 MQTT_SCHEMA.md §3.4 |
envelope 생성/파싱은 packages/topics/의 공용 wrapper(wrap())를 사용한다 — 토픽 문자열·스키마를 서비스 코드에 inline으로 재정의하지 않는다 (프로젝트 invariant).
robot_id는 string 타입 + config/robots.yaml 레지스트리 런타임 검증이다 — literal union("spot-01" | ...)으로 타입을 좁히지 않는다.
publish(topic, payload, qos, retain) 네 인자를 지정한다. QoS·retain은 토픽별로 정해진 값을 그대로 쓴다:
| adapter가 발행하는 토픽 | type | QoS | retain | 비고 |
|---|---|---|---|---|
robot/{robot_id}/state |
robot.state |
0 | true | 1Hz 텔레메트리. pose(world 좌표)·battery·current_action·faults·connection |
robot/{robot_id}/ack |
robot.ack |
1 | false | 같은 command_id로 accepted → started → completed/failed/rejected 순차 발행 |
robot/{robot_id}/fault |
robot.fault |
1 | true | 해제 시 동일 토픽에 active: false를 retain false로 발행 |
safety/estop |
safety.estop |
2 | true | 자체 감지 시에만 |
system/heartbeat/robot-adapter |
system.heartbeat |
0 | true | 라이브니스 |
# 1Hz 텔레메트리 발행
state = wrap("robot.state", telemetry.snapshot(),
source="/services/mcp-robot-adapter", subject="robot/spot-01")
client.publish(
topic="robot/spot-01/state", # 어디로 — 구체적 topic (와일드카드 불가)
payload=json.dumps(state), # 무엇을 — envelope JSON 문자열
qos=0, # 어떤 보장으로 — state는 0 (고빈도, 유실 허용)
retain=True, # 보관 여부 — 새 구독자가 즉시 최신 상태 획득
)그리고 정상 종료 시에도 offline state를 직접 발행하고 끊는다 — 그래야 broker가 유언장을 발동하지 않고, retain 상태도 깨끗하게 offline으로 남는다:
def shutdown_handler(*_):
client.publish("robot/spot-01/state", json.dumps(will), qos=1, retain=True)
client.disconnect()publish는 성공했는데 (ACL 때문에) broker가 버렸을 수도 있다 — 발행 코드만 믿지 말고 반드시 구독 터미널로 확인한다. 방법은 §6, HA 엔티티까지의 확인은 §7.
§1.2에서 설명했듯 MQTT 자체에는 topic 생성 절차가 없다 — 이 프로젝트에서 "토픽 등록"은 문서·코드·권한을 맞추는 일이다. 단일 소스는 두 곳: 문서는 MQTT_SCHEMA.md, 코드는 packages/topics/ (SERVICES.md §9.2). 어느 쪽도 거치지 않고 서비스 코드에 토픽 문자열을 inline으로 박는 것이 가장 흔한 결함이다.
| 단계 | 작업 | 위치 |
|---|---|---|
| 1 |
토픽 카탈로그에 정의 — 토픽명, type, 발행자/구독자, QoS, retain, payload 인터페이스, JSON 예시 |
MQTT_SCHEMA.md §3 + §4 발행자/구독자 매트릭스 |
| 2 | 스키마 코드 반영 — 상수·Pydantic 모델 추가 후 codegen으로 TS(Zod)↔Python 동기화 | packages/topics/ |
| 3 |
ACL 갱신 — 발행자에 topic write, 구독자에 topic read 추가 |
mosquitto/config/acl |
| 4 | broker 적용 — 재시작 없이 reload | docker kill -s HUP mosquitto |
| 5 | retain=true 토픽이면 LWT 정책 추가 | MQTT_SCHEMA.md §4.5 표 |
⚠️ ACL 갱신을 빠뜨리면 broker가 publish를 에러 없이 조용히 drop한다. "발행했는데 구독자에 안 보임" → 가장 먼저 ACL을 의심할 것.
MQTT_SCHEMA.md §7 그대로:
- 새 필드는 optional로 추가 — 기존 필드 삭제·타입 변경 금지 (backward 호환).
-
breaking change는
type버전화:robot.state.v2처럼.v2suffix → v1·v2 병행 발행 → 모든 구독자 마이그레이션 후 v1 폐기. - QoS·retain 변경도 스키마 변경으로 취급 — MQTT_SCHEMA.md §2 표를 함께 갱신.
전부 broker 컨테이너에 들어 있는 명령줄 도구 mosquitto_sub(구독) / mosquitto_pub(발행) 으로 가능하다 — 별도 도구 설치 불필요. 인증이 켜져 있으므로 -u(username) / -P(password)는 항상 필요하다.
테스트의 기본 자세: 구독 터미널을 먼저 열어 두고, 그 다음 발행한다. (retain이 아닌 메시지는 발행 순간의 구독자에게만 가므로, 순서가 반대면 아무것도 못 본다.)
# 터미널 1: 모든 토픽 라이브 테일 (-t '#': 전체 구독, -v: 토픽명도 같이 출력)
docker exec -it mosquitto mosquitto_sub -h localhost -u robot-adapter -P '<pwd>' -t '#' -v
# robot 도메인만
docker exec -it mosquitto mosquitto_sub -h localhost -u robot-adapter -P '<pwd>' \
-t 'robot/#' -v
# retained 메시지만 확인 (현재 로봇 상태 스냅샷)
docker exec -it mosquitto mosquitto_sub -h localhost -u robot-adapter -P '<pwd>' \
-t 'robot/+/state' -v --retained-only
# 특정 trace_id 추적
docker exec -it mosquitto mosquitto_sub -h localhost -u robot-adapter -P '<pwd>' -t '#' -v \
| grep '0af7651916cd43dd8448eb211c80319c'HA 폴백 경로를 흉내내 robot/{robot_id}/command를 발행한다. homeassistant 계정을 사용해야 한다 — robot-adapter는 이 토픽의 write 권한이 없다 (ACL이 의도대로 동작하는지의 검증이기도 하다).
docker exec -it mosquitto mosquitto_pub -h localhost -u homeassistant -P '<pwd>' \
-t 'robot/spot-01/command' -q 1 -m '{
"specversion": "1.0", "type": "robot.command",
"source": "/services/home-assistant", "id": "01HW9X...",
"time": "2026-05-04T10:00:00Z", "datacontenttype": "application/json",
"data": {
"command_id": "01HW9X...", "robot_id": "spot-01",
"capability": "look_at", "params": { "target_session_id": "01HW8M..." },
"priority": "operator",
"source": { "service": "home-assistant", "trace_id": "manual-test" },
"expires_at": "2026-05-04T10:00:05Z"
}
}'기대 결과 — 터미널 1(robot/# 구독)에 adapter가 발행한 ack 시퀀스가 보인다:
robot/spot-01/ack {... "data": {"command_id": "01HW9X...", "status": "accepted", ...}}
robot/spot-01/ack {... "status": "started" ...}
robot/spot-01/ack {... "status": "completed" ...}
rejected가 오면 reason 필드에 safety interlock 거부 사유(estop/fence/proximity/watchdog)가 들어 있다. ACL 위반이면 아무것도 안 온다 (broker silent drop).
테스트 중 만든 retained 메시지는 빈(null) payload를 retain으로 발행하면 지워진다 (MQTT의 retain 삭제 규약):
docker exec mosquitto mosquitto_pub -h localhost -u robot-adapter -P '<pwd>' \
-t 'robot/spot-01/state' -r -n # -r: retain, -n: null payload → broker가 보관분 삭제HA 쪽에서도 같은 테스트를 할 수 있다 — 개발자 도구 → 액션에서 mqtt.publish 호출 (운영자 대시보드의 수동 명령 버튼이 쓰는 것과 같은 경로다).
adapter가 robot/{robot_id}/state를 발행하면 HA 엔티티(§3.2)가 갱신된다. 확인 방법은 세 가지.
HA 접속: 로컬 PC 브라우저에서 http://<로컬PC>:8123 (컨테이너 내부에서는 http://host.docker.internal:8123).
-
개발자 도구 → 상태:
sensor.spot_01_battery검색 → 현재 값·attributes·last_updated 확인. 발행 직후 값이 바뀌는지 본다. -
개발자 도구 → MQTT → "Listen to a topic":
robot/+/state입력 → HA가 실제 수신하는 raw payload를 그대로 본다. 엔티티는 안 바뀌는데 여기엔 보인다면value_template문제, 여기도 안 보이면 broker/ACL 문제로 구분할 수 있다. -
기록(History): 엔티티 클릭 → 그래프. recorder가 SQLite에 자동 기록한다 (
purge_keep_days: 3).
HA_TOKEN(long-lived access token, 발급 절차는 DEPLOYMENT.md §9.2)으로 조회한다:
curl -s -H "Authorization: Bearer $HA_TOKEN" \
http://host.docker.internal:8123/api/states/sensor.spot_01_battery | jq{
"entity_id": "sensor.spot_01_battery",
"state": "87",
"attributes": { "unit_of_measurement": "%", "friendly_name": "SPOT-01 Battery" },
"last_updated": "2026-05-04T10:00:01.456Z"
}orchestrator가 컨텍스트 조회에 쓰는 것과 동일한 경로이므로, 여기서 보이는 값이 곧 LLM이 보는 값이다.
- 터미널에서
mosquitto_sub -t 'robot/+/state'로 발행 자체 확인 (§6.1) - HA 개발자 도구 → MQTT listen으로 HA 수신 확인
- 개발자 도구 → 상태에서 엔티티 값 갱신 확인
-
curl /api/states/...로 REST 조회 확인 — 여기까지 통과하면 orchestrator·운영자 대시보드 모두에서 보인다
흔한 함정: discovery config나
packages/robots.yaml의value_template이 envelope을 고려하지 않은 경우. payload는 CloudEvents envelope으로 감싸여 있으므로{{ value_json.battery_pct }}가 아니라 **{{ value_json.data.battery_pct }}**다.
-
mosquitto_passwd로robot-adapter등록,passwd권한1883:1883 / 0600, broker reload -
acl에 robot-adapter 블록 확인 (신규 토픽은 §5 절차: MQTT_SCHEMA.md →packages/topics→ ACL → reload) -
deploy/.env.secret에MOSQUITTO_PWD_ROBOT_ADAPTER,SPOT_USERNAME/PASSWORD기입 - SPOT service account는 dedicated 계정(
zerone-spot-svc, Operator+EStop, Admin 금지) — DEPLOYMENT.md §9.2 - connect 전 LWT 등록 + graceful shutdown에서 offline state 발행 (§4 STEP 3·7)
- 모든 발행 메시지 CloudEvents envelope +
packages/topics/wrapper 사용 (§4 STEP 6) -
robot/+/command수신분이 MCP 경로와 같은 큐·interlock으로 수렴하는지 확인 - HA에서
sensor.spot_01_battery등 엔티티가 보이는지 확인 —value_template은value_json.data.*(§7.3 E2E 한 바퀴) -
safety/estop수신 시 큐 fence + 운영자 reset까지 유지 (safety > operator > llm)
서비스 전체 책임·디렉터리 구조·테스트 전략은 SERVICES.md §6 참고.