16 KiB
16 KiB
🔌 Multi-Agent Mux (MAM): 신규 에이전트 타입 추가 가이드 (New Agent Integration Guide)
본 문서는 Multi-Agent Mux (MAM) 프레임워크에 새로운 AI 에이전트 CLI/TUI 백엔드(예: grok, codex, opencode, kimi, cursor 등)를 추가하기 위한 아키텍처 구조와 5단계 필수 작업 절차를 안내합니다.
1. 아키텍처 개요 (Architecture Overview)
MAM은 에이전트별 동작 특성(TUI 프롬프트 패턴, 세션 복원 인자, 아티팩트 저장소 등)을 Python 기반의 **어댑터 패턴(BaseAgentAdapter)**으로 격리하여 관리합니다. 코어 런타임(Shell, Herdr Daemon, MQTT 브로커)을 수정할 필요 없이 어댑터를 플러그인 형태로 추가할 수 있습니다.
┌──────────────────────────────────────────────────────────┐
│ Herdr Runtime │
│ (herdr agent start <name> --kind <kind> -- <cmd>) │
└────────────────────────────┬─────────────────────────────┘
│
▼
┌──────────────────────────────────────────────────────────┐
│ lib.sh (Shell Runtime) │
│ • resolve_agent_type_from_registry() │
│ • wait_for_tui_ready() (via MAM_READY_TOKENS) │
│ • send_keys_safe() / stop_session.sh │
└────────────────────────────┬─────────────────────────────┘
│
eval $(python -m lib_py.agents facts <agent>)
│
▼
┌──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ lib_py.agents Framework │
│ │
│ ┌──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐ │
│ │ BaseAgentAdapter │ │
│ │ • name, own_key, ready_tokens, exit_key, delegate_agent_key, identity_cache_fields │ │
│ │ • input_prompt, input_placeholder, input_rule_pattern │ │
│ │ • artifact_path(), verify_artifact(), purge_artifacts() │ │
│ │ • spawn_spec(), resume_spec(), auth_ok(), discover() │ │
│ └────────────────────────────────────────────────────────────────┬─────────────────────────────────────────────────────────────────┘ │
│ │ │
│ ┌────────────────────┬───────────────────────────────┼───────────────────────────────┬─────────────────────┐ │
│ ▼ ▼ ▼ ▼ ▼ │
│ ClaudeAgentAdapter AgyAgentAdapter ClineAgentAdapter HermesAgentAdapter [NewAgentAdapter] │
│ (Claude Code) (Antigravity) (Cline) (Hermes) (Grok / Codex / ...) │
└──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘
2. 5단계 필수 구현 가이드 (Step-by-Step Implementation)
신규 에이전트 <agent>(예: grok)를 연동하기 위해서는 다음 5개 파일/영역의 수정이 필요합니다.
📁 .agents/skills/lib_py/agents/adapters/<agent>.py # Step 1: 어댑터 클래스 구현
📁 .agents/skills/lib_py/agents/registry.py # Step 2: 어댑터 레지스트리 등록
📁 .agents/skills/lib.sh # Step 3: 쉘 디스패치 및 kind 매핑
📁 Herdr Runtime Configuration # Step 4: Herdr CLI 데몬 kind 호환성 확인
📁 tests/test_a4_adapter_contract.py # Step 5: 어댑터 계약 단위 테스트 작성
Step 1: Python 어댑터 구현 (.agents/skills/lib_py/agents/adapters/<agent>.py)
BaseAgentAdapter를 상속하는 <Agent>AgentAdapter 클래스를 생성합니다.
"""
<agent>.py — <Agent> agent adapter for Multi-Agent Mux (MAM)
"""
import os
from typing import Any, List, Optional
from ..base import BaseAgentAdapter, DiscoveryContext
class GrokAgentAdapter(BaseAgentAdapter):
@property
def name(self) -> str:
"""에이전트 고유 식별자 (소문자 영문)"""
return 'grok'
@property
def own_key(self) -> str:
"""YAML (.mam/agent-sessions.yaml) 세션 상태에 저장될 키 이름"""
return 'grok_session_id_own'
@property
def ready_tokens(self) -> str:
"""TUI 부팅 완료 및 사용자 입력 대기 상태를 감지하는 정규식 패턴"""
return r'Grok|xAI|Assistant|❯|>>>'
@property
def exit_key(self) -> str:
"""정상 종료(Graceful exit) 시 TUI에 입력할 키/명령어"""
return '/exit' # 또는 'Exit', 'quit', ':q'
@property
def delegate_agent_key(self) -> str:
"""delegate-job MQTT 이벤트 채널에서 사용할 수신자 식별자"""
return 'grok-cli'
@property
def identity_cache_fields(self) -> tuple:
"""세션 복원 시 캐시할 식별자 필드 튜플"""
return ('session_id',)
# =========================================================================
# TUI 프롬프트 영역 비주얼 파싱 (선택적 커스터마이징)
# =========================================================================
@property
def input_prompt(self) -> str:
return '❯'
@property
def input_placeholder(self) -> str:
return ''
@property
def input_rule_pattern(self) -> str:
return r'─{10,}'
# =========================================================================
# 아티팩트 및 대화 히스토리 수명 주기 관리
# =========================================================================
def artifact_path(self, uuid: str, ctx: DiscoveryContext) -> str:
"""해당 세션 UUID의 대화 로그/아티팩트가 저장되는 절대 경로 반환"""
return f"{ctx.home_dir}/.grok/sessions/{uuid}.json"
def verify_artifact(self, uuid: str, ctx: DiscoveryContext) -> bool:
"""디스크 상의 아티팩트 유효성 및 타임스탬프 검증"""
path = self.artifact_path(uuid, ctx)
if not os.path.exists(path):
return False
if ctx.epoch and os.path.getmtime(path) < ctx.epoch:
return False
return True
def purge_artifacts(self, uuid: str, ctx: DiscoveryContext) -> List[str]:
"""--purge-conversation 옵션 호출 시 디스크의 대화 파일 삭제"""
path = self.artifact_path(uuid, ctx)
if os.path.exists(path):
os.remove(path)
return [path]
return []
# =========================================================================
# CLI 실행 및 세션 복원 인자 생성
# =========================================================================
def spawn_spec(self, binary: str, session_uuid: str = "", use_wrapper: bool = False) -> str:
"""신규 세션 기동 시 실행할 커맨드라인 문자열 생성"""
if session_uuid:
return f"{binary} --session {session_uuid}"
return binary
def resume_spec(self, binary: str, session_uuid: str, materialized: bool = False) -> str:
"""기존 세션 복원(Resume) 시 실행할 커맨드라인 문자열 생성"""
if materialized and session_uuid:
return f"{binary} --resume {session_uuid}"
return f"{binary} --session {session_uuid}" if session_uuid else binary
# =========================================================================
# 인증 상태 검사 및 세션 디스커버리
# =========================================================================
def auth_ok(self, run_cmd: Optional[Any] = None) -> bool:
"""에이전트 실행에 필요한 API 키/토큰/설정 파일 존재 여부 확인"""
token_file = os.path.expanduser('~/.grok/token.json')
return os.path.exists(token_file) or bool(os.environ.get('GROK_API_KEY') or os.environ.get('XAI_API_KEY'))
def discover(self, ctx: DiscoveryContext) -> List[str]:
"""디스크의 세션 저장소에서 워크스페이스와 일치하는 세션 UUID 목록 탐색"""
return []
Step 2: 어댑터 레지스트리 등록 (.agents/skills/lib_py/agents/registry.py)
registry.py의 _ADAPTERS 딕셔너리에 새 어댑터 인스턴스를 등록합니다.
# 1. 어댑터 임포트
from .adapters.grok import GrokAgentAdapter
# 2. 레지스트리 맵에 등록
_ADAPTERS: Dict[str, BaseAgentAdapter] = {
'claude': ClaudeAgentAdapter(),
'agy': AgyAgentAdapter(),
'hermes': HermesAgentAdapter(),
'cline': ClineAgentAdapter(),
'grok': GrokAgentAdapter(), # 👈 추가
}
등록 후 즉시 사용 가능한 CLI 명령:
python -m lib_py.agents facts grok➔ 쉘 환경변수(MAM_READY_TOKENS,MAM_EXIT_KEY등) 자동 출력python -m lib_py.agents spawn-spec grok grok➔ 기동 커맨드 생성python -m lib_py.agents resume-spec grok grok <uuid>➔ 복원 커맨드 생성python -m lib_py.agents resolve <session_name>➔ 세션 이름 기반 에이전트 자동 식별
Step 3: 쉘 런타임 디스패치 연동 (.agents/skills/lib.sh)
lib.sh에서 Herdr 세션 기동 시 올바른 kind가 전달되도록 패턴 매칭을 추가합니다.
1) Herdr Session Kind 매핑 (lib.sh:357-371)
case "$name" in
*-creator-claude|*-planner-claude|*-reviewer-claude) kind="claude" ;;
*-creator-agy|*-planner-agy|*-reviewer-agy) kind="agy" ;;
*-creator-hermes|*-planner-hermes|*-reviewer-hermes) kind="hermes" ;;
*-creator-cline|*-planner-cline|*-reviewer-cline) kind="cline" ;;
*-creator-grok|*-planner-grok|*-reviewer-grok) kind="grok" ;; # 👈 추가
*)
if echo "$name" | grep -qi "claude"; then kind="claude"
elif echo "$name" | grep -qi "agy"; then kind="agy"
elif echo "$name" | grep -qi "hermes"; then kind="hermes"
elif echo "$name" | grep -qi "cline"; then kind="cline"
elif echo "$name" | grep -qi "grok"; then kind="grok" # 👈 추가
else kind="generic"; fi
;;
esac
2) 바이너리 이름 중복 제거 튜플 (lib.sh:385)
herdr agent start 호출 시 첫 번째 인자로 전달되는 바이너리 중복을 방지하기 위해 등록합니다:
if [[ " claude agy hermes cline grok " =~ " ${cmd_binary} " ]]; then
Step 4: Herdr CLI 데몬 호환성 확인
- 내장 kind 지원 여부 확인:
herdr데몬이 해당--kind를 자체적으로 파싱하는지 확인합니다.- 별도 내장 파서가 없는 경우
--kind generic으로도 세션 기동 및 터미널 I/O 제어가 완전하게 지원됩니다.
Step 5: 테스트 작성 및 계약 검증 (tests/test_a4_adapter_contract.py)
신규 어댑터가 MAM 계약 규격을 완벽하게 충족하는지 검증하는 단위 테스트를 등록합니다.
# tests/test_a4_adapter_contract.py
def test_agent_adapter_registry():
"""모든 등록된 에이전트 어댑터 인스턴스 검증"""
adapters = get_all_adapters()
assert set(adapters.keys()) == {'claude', 'agy', 'hermes', 'cline', 'grok'}
def test_adapter_required_properties():
"""어댑터 필수 프로퍼티 무결성 검증"""
# ('grok', ('grok_session_id_own', r'Grok|xAI|Assistant|❯|>>>', '/exit', 'grok-cli', ('session_id',)))
테스트 실행:
.venv/bin/python -m pytest tests/test_a4_adapter_contract.py -v
3. 에이전트 백엔드별 구현 난이도 매트릭스
| 에이전트 백엔드 | 세션 저장소 형식 | 인증 방식 | 프롬프트 감지 난이도 | 난이도 Tier | 예상 소요 공수 |
|---|---|---|---|---|---|
| OpenCode | ~/.opencode/sessions/*.json |
로컬 토큰 / API Key | 쉬움 (OpenCode|Chat) |
Tier 1 (낮음) | ~0.5일 |
| Codex CLI | ~/.codex/projects/*.jsonl |
OPENAI_API_KEY |
쉬움 (Codex|❯) |
Tier 1 (낮음) | ~0.5일 |
| Kimi CLI | ~/.kimi/history.db (SQLite) |
~/.kimi/config |
쉬움 (Moonshot|Kimi) |
Tier 1 (낮음) | ~0.5일 |
| Grok-Build | ~/.grok/sessions/*.json |
토큰 파일 / Env | 쉬움 (Grok|Building) |
Tier 1 (낮음) | ~0.5일 |
| Cursor CLI | ~/.cursor/ (Headless/RPC) |
OAuth / 쿠키 | 보통 (RPC 포트/상태 파싱) | Tier 2 (중간) | ~1.5일 |
| Local LLM | ollama / vllm CLI stdout |
로컬 호스트 | 보통 (>>> ANSI 패턴 감지) |
Tier 2 (중간) | ~1.0일 |
4. 완료 정의 (Definition of Done Checklist)
신규 에이전트 연동 PR 또는 커밋 전 다음 항목을 확인합니다:
.agents/skills/lib_py/agents/adapters/<agent>.py에BaseAgentAdapter모든 추상 메서드/프로퍼티 구현 완료.agents/skills/lib_py/agents/registry.py에 어댑터 등록 완료.agents/skills/lib.sh에kind매핑 및 바이너리 패턴 등록 완료tests/test_a4_adapter_contract.py테스트 케이스 추가 및 통과tests/test_tier1_unit.py및 전체 테스트 스위트 통과 (pytest tests/)- 세션 기동(
multi-agent-mux-create), 정지(multi-agent-mux-stop), 복원(multi-agent-mux-resume) 동작 검증 완료