diff --git a/.agent/references/known-issues.md b/.agent/references/known-issues.md index d74a1ba..c45df39 100644 --- a/.agent/references/known-issues.md +++ b/.agent/references/known-issues.md @@ -57,3 +57,9 @@ - **원인**: 단일 검색 전략, suffix(ASW HEVC) 항상 부착 - **해결**: 6단계 fallback 전략 (romaji±suffix → 원제±suffix → 한글±suffix) - **주의**: 외부 API 검색 시 반드시 다중 전략 + suffix 토글 구현 + +### [2026-03-12] Gemini CLI MCP — settings.json 위치 +- **증상**: MCP 도구가 인식되지 않음 (프로젝트 .gemini/settings.json에 설정했으나 실패) +- **원인**: Gemini CLI는 `cwd` 기준으로 `.gemini/settings.json`을 탐색. cwd가 다른 워크스페이스면 MCP 설정 못 찾음 +- **해결**: `~/.gemini/settings.json` (홈 레벨)에 mcpServers 등록. `_set_thinking_budget`에서 자동 관리 +- **주의**: MCP 서버 설정은 반드시 홈 레벨 settings.json에 등록. 프로젝트 레벨은 불충분 diff --git a/.gemini/settings.json b/.gemini/settings.json new file mode 100644 index 0000000..9482b82 --- /dev/null +++ b/.gemini/settings.json @@ -0,0 +1,16 @@ +{ + "mcpServers": { + "anime": { + "command": "C:\\ProgramData\\miniforge3\\envs\\variet-agent\\python.exe", + "args": ["mcp_servers/anime_server.py"], + "cwd": "C:\\Users\\Variet-Worker\\Desktop\\variet-agent", + "trust": true + }, + "infra": { + "command": "C:\\ProgramData\\miniforge3\\envs\\variet-agent\\python.exe", + "args": ["mcp_servers/infra_server.py"], + "cwd": "C:\\Users\\Variet-Worker\\Desktop\\variet-agent", + "trust": true + } + } +} diff --git a/api/discord_bot.py b/api/discord_bot.py index a9832a6..ff075e1 100644 --- a/api/discord_bot.py +++ b/api/discord_bot.py @@ -124,52 +124,27 @@ async def safe_send_embed(channel, embed: discord.Embed): # ────────────────────────────────────────────── -# 통합 프롬프트 (1회 호출: 분류 + 응답/계획) +# 에이전트 호출 (MCP 도구 자동 사용) # ────────────────────────────────────────────── -async def _unified_call(text: str, history: str, project_path: str) -> dict: - """통합 프롬프트로 1회 호출 — chat/task/clarify 자동 분기.""" - gemini = GeminiCaller(project_path) +async def _agent_call(text: str, history: str, project_path: str) -> str: + """Gemini CLI 에이전트 모드로 호출 — MCP 도구를 자율적으로 사용. - # docs 인덱스 주입 - from core.docs_manager import DocsManager - docs = DocsManager(project_path) - docs_index = docs.get_docs_index() + 분류(chat/task/anime) 없이 에이전트가 직접 판단하여 + 도구를 사용하거나 바로 답변합니다. + """ + gemini = GeminiCaller(project_path) context = ( f"{history}" f"## Workspace\nPath: {project_path}\n\n" - f"## Project Docs\n{docs_index}\n\n" f"## User Message\n{text}" ) - raw = await gemini.call("unified", context, timeout=120) - - # JSON 추출 - try: - # ```json ... ``` 패턴 - match = re.search(r'```json\s*\n(.*?)\n\s*```', raw, re.DOTALL) - if match: - return json.loads(match.group(1)) - - # 중첩 { } 찾기 - brace_depth = 0 - start = -1 - for i, ch in enumerate(raw): - if ch == '{': - if brace_depth == 0: - start = i - brace_depth += 1 - elif ch == '}': - brace_depth -= 1 - if brace_depth == 0 and start >= 0: - return json.loads(raw[start:i + 1]) - except (json.JSONDecodeError, AttributeError): - pass - - # 파싱 실패 → chat으로 처리 - logger.warning(f"통합 프롬프트 JSON 파싱 실패: {raw[:100]}") - return {"mode": "chat", "response": raw} + response = await gemini.call_agent( + "agent", context, cwd=project_path, timeout=300, + ) + return response # ────────────────────────────────────────────── @@ -297,75 +272,68 @@ async def on_message(message: discord.Message): await message.reply("실행 중인 작업이 없습니다.") return - # 통합 프롬프트 호출 - async with message.channel.typing(): + # 에이전트 호출 (MCP 도구 자동 사용) + channel_id = message.channel.id + if channel_id in _running_tasks and not _running_tasks[channel_id].done(): + await message.reply("⚠️ 이미 작업이 실행 중입니다. `취소` 후 다시 요청하세요.") + return + + async def _tracked_agent(): + progress_msg = None try: - history = await _get_channel_history(message.channel, limit=10) - result = await _unified_call(user_text, history, ws.path) - except GeminiCallError as e: - await message.reply(f"⚠️ AI 호출 오류: {str(e)[:200]}") - return - except Exception as e: - logger.error(f"통합 호출 오류: {e}", exc_info=True) - await message.reply(f"❌ 오류: {str(e)[:200]}") - return - - mode = result.get("mode", "chat") - logger.info(f"통합 분류: {mode} - \"{user_text[:50]}\"") - - if mode == "anime": - # 애니메이션 도구 자동 호출 - async with message.channel.typing(): - await _handle_anime(message, result) - elif mode == "task": - # Git/Vikunja 미설정 안내 (차단하지 않음) - if not ws.is_ready: - missing = ws.missing_configs - note = " / ".join(missing) - await message.channel.send( - f"ℹ️ {note} 미설정 상태입니다. 로컬 작업만 진행됩니다." + # 진행 표시 + progress_msg = await message.channel.send( + embed=discord.Embed( + title="🤖 에이전트 처리 중...", + description=f"```{user_text[:200]}```", + color=0xF39C12, + ) ) - # 작업을 추적 가능한 Task로 실행 - channel_id = message.channel.id - if channel_id in _running_tasks and not _running_tasks[channel_id].done(): - await message.reply("⚠️ 이미 작업이 실행 중입니다. `취소` 후 다시 요청하세요.") - return + async with message.channel.typing(): + history = await _get_channel_history(message.channel, limit=10) + response = await _agent_call(user_text, history, ws.path) - async def _tracked_task(): - try: - await _handle_task(message, user_text, ws) - except asyncio.CancelledError: - await message.channel.send( - embed=discord.Embed( - title="🛑 작업 취소됨", - description="작업이 사용자에 의해 취소되었습니다.", - color=0xE74C3C, - ) + logger.info(f"에이전트 응답: \"{user_text[:50]}\" -> {len(response)}자") + + # 진행 메시지 삭제 + if progress_msg: + try: + await progress_msg.delete() + except Exception: + pass + + if not response: + await message.reply("응답을 생성하지 못했습니다.") + return + + # 응답 전송 + if len(response) <= 2000: + await message.reply(response) + else: + for i in range(0, len(response), 4000): + chunk = response[i:i + 4000] + embed = discord.Embed(description=chunk, color=0x3498DB) + await message.channel.send(embed=embed) + + except asyncio.CancelledError: + await message.channel.send( + embed=discord.Embed( + title="🛑 작업 취소됨", + description="작업이 사용자에 의해 취소되었습니다.", + color=0xE74C3C, ) - finally: - _running_tasks.pop(channel_id, None) + ) + except GeminiCallError as e: + await message.reply(f"⚠️ AI 호출 오류: {str(e)[:200]}") + except Exception as e: + logger.error(f"에이전트 호출 오류: {e}", exc_info=True) + await message.reply(f"❌ 오류: {str(e)[:200]}") + finally: + _running_tasks.pop(channel_id, None) - task = asyncio.create_task(_tracked_task()) - _running_tasks[channel_id] = task - elif mode == "clarify": - question = result.get("question", "더 구체적으로 말씀해 주시겠어요?") - embed = discord.Embed( - title="🤔 확인이 필요해요", - description=question, - color=0xF39C12, - ) - await message.reply(embed=embed) - else: - # chat — 즉답 - response = result.get("response", "응답을 생성하지 못했습니다.") - if len(response) <= 2000: - await message.reply(response) - else: - for i in range(0, len(response), 4000): - chunk = response[i:i + 4000] - embed = discord.Embed(description=chunk, color=0x3498DB) - await message.channel.send(embed=embed) + task = asyncio.create_task(_tracked_agent()) + _running_tasks[channel_id] = task # ────────────────────────────────────────────── @@ -1325,13 +1293,11 @@ async def _create_task_thread( try: async with thread.typing(): history = "" - result = await _unified_call(request_text, history, ws.path) + response = await _agent_call(request_text, history, ws.path) - mode = result.get("mode", "chat") - logger.info(f"[스레드] 통합 분류: {mode} - \"{request_text[:50]}\"") + logger.info(f"[스레드] 에이전트 응답: \"{request_text[:50]}\" -> {len(response)}자") - if mode == "chat": - response = result.get("response", "응답을 생성하지 못했습니다.") + if response: if len(response) <= 2000: await thread.send(response) else: @@ -1339,23 +1305,8 @@ async def _create_task_thread( chunk = response[i:i + 4000] embed = discord.Embed(description=chunk, color=0x3498DB) await thread.send(embed=embed) - elif mode == "clarify": - question = result.get("question", "더 구체적으로 말씀해 주시겠어요?") - embed = discord.Embed( - title="🤔 확인이 필요해요", - description=question, - color=0xF39C12, - ) - await thread.send(embed=embed) - else: # task - # task 모드 — 스레드에서 파이프라인 안내 - await thread.send( - embed=discord.Embed( - title="⚙️ 작업 모드 감지", - description="이 스레드에서 작업 요청을 다시 입력해주세요.\n스레드 내 메시지는 자동으로 처리됩니다.", - color=0xF39C12, - ) - ) + else: + await thread.send("응답을 생성하지 못했습니다.") except Exception as e: logger.error(f"스레드 초기 호출 오류: {e}", exc_info=True) await thread.send(f"⚠️ 초기 호출 오류: {str(e)[:200]}") diff --git a/core/gemini_caller.py b/core/gemini_caller.py index 808e818..937df3a 100644 --- a/core/gemini_caller.py +++ b/core/gemini_caller.py @@ -14,12 +14,14 @@ from pathlib import Path logger = logging.getLogger("variet.gemini") ROLE_PROMPTS_DIR = Path(__file__).parent.parent / "prompts" +PROJECT_ROOT = Path(__file__).parent.parent GEMINI_MODEL = "gemini-3-flash-preview" # 역할별 thinkingBudget (토큰 단위) # 512=가벼운 분류/요약, 4096=계획/검수, 8192=구현/비평 ROLE_THINKING: dict[str, int] = { "unified": 512, + "agent": 4096, "summarizer": 512, "planner": 4096, "coder": 8192, @@ -57,28 +59,50 @@ class GeminiCaller: "--approval-mode", "yolo"] return ["gemini", "--model", GEMINI_MODEL, "--approval-mode", "yolo"] + # MCP 서버 설정 (홈 레벨에 등록) + _MCP_SERVERS = { + "anime": { + "command": str(Path(sys.executable)), + "args": [str(PROJECT_ROOT / "mcp_servers" / "anime_server.py")], + "cwd": str(PROJECT_ROOT), + "trust": True, + }, + "infra": { + "command": str(Path(sys.executable)), + "args": [str(PROJECT_ROOT / "mcp_servers" / "infra_server.py")], + "cwd": str(PROJECT_ROOT), + "trust": True, + }, + } + def _set_thinking_budget(self, role: str): - """역할별 thinkingBudget을 settings.json에 반영.""" + """역할별 thinkingBudget + MCP 서버 설정을 settings.json에 반영.""" budget = ROLE_THINKING.get(role, DEFAULT_THINKING) try: if _SETTINGS_PATH.exists(): settings = json.loads(_SETTINGS_PATH.read_text(encoding="utf-8")) else: + _SETTINGS_PATH.parent.mkdir(parents=True, exist_ok=True) settings = {} - # modelConfigs.default.thinkingConfig.thinkingBudget 설정 + # thinkingBudget configs = settings.setdefault("modelConfigs", {}) default = configs.setdefault("default", {}) thinking = default.setdefault("thinkingConfig", {}) thinking["thinkingBudget"] = budget + # MCP 서버 (홈 레벨에 등록 — cwd와 무관하게 항상 사용 가능) + mcp_servers = settings.setdefault("mcpServers", {}) + for name, config in self._MCP_SERVERS.items(): + mcp_servers[name] = config + _SETTINGS_PATH.write_text( json.dumps(settings, indent=2, ensure_ascii=False), encoding="utf-8", ) logger.debug(f"thinkingBudget={budget} for role={role}") except Exception as e: - logger.warning(f"thinkingBudget 설정 실패 (role={role}): {e}") + logger.warning(f"settings.json 업데이트 실패 (role={role}): {e}") def _clean_output(self, raw: str) -> str: """Gemini 출력에서 노이즈 라인 제거.""" @@ -190,16 +214,30 @@ class GeminiCaller: else: system_prompt = f"You are a {role}. Respond in Korean." + # 역할에 따라 다른 지시 + if role == "agent": + # 범용 에이전트: 도구 사용 여부를 자율 판단 + footer = ( + f"프로젝트 루트: {cwd}\n" + f"모든 응답은 한국어로 작성하세요.\n" + f"파일 작업이 필요하면 직접 수행하고, 대화만 필요하면 바로 답변하세요." + ) + else: + # coder 등 파일 작업 전용 역할 + footer = ( + f"프로젝트 루트: {cwd}\n" + f"파일을 직접 생성/수정하세요. 코드블록으로 출력하지 말고, 실제 파일로 저장하세요.\n" + f"작업 완료 후 변경한 파일 목록을 간단히 출력하세요.\n" + f"모든 응답, 주석, 문서는 반드시 한국어로 작성하세요." + ) + full_input = ( f"=== SYSTEM INSTRUCTIONS ===\n" f"{system_prompt}\n\n" f"=== TASK ===\n" f"{context}\n\n" f"=== IMPORTANT ===\n" - f"프로젝트 루트: {cwd}\n" - f"파일을 직접 생성/수정하세요. 코드블록으로 출력하지 말고, 실제 파일로 저장하세요.\n" - f"작업 완료 후 변경한 파일 목록을 간단히 출력하세요.\n" - f"모든 응답, 주석, 문서는 반드시 한국어로 작성하세요." + f"{footer}" ) try: diff --git a/docs/devlog/2026-03-12.md b/docs/devlog/2026-03-12.md new file mode 100644 index 0000000..5f896da --- /dev/null +++ b/docs/devlog/2026-03-12.md @@ -0,0 +1,5 @@ +# 2026-03-12 개발 로그 + +| # | 시간 | 작업 | 커밋 | 상태 | +|---|------|------|------|------| +| 001 | 16:45 | MCP 기반 에이전트 아키텍처 재설계 — unified.md 분류기 → Gemini CLI + MCP 자율 에이전트 전환 | `pending` | ✅ | diff --git a/docs/devlog/entries/20260312-001.md b/docs/devlog/entries/20260312-001.md new file mode 100644 index 0000000..e33116b --- /dev/null +++ b/docs/devlog/entries/20260312-001.md @@ -0,0 +1,27 @@ +# MCP 기반 에이전트 아키텍처 재설계 + +- **시간**: 2026-03-12 15:00~16:45 +- **Commit**: `pending` +- **Vikunja**: 신규 태스크 + +## 결정 사항 + +- Gemini CLI의 네이티브 MCP 지원을 활용하여 에이전트 루프를 직접 구현하지 않고 CLI에 위임 +- unified.md 단발성 JSON 분류 → Gemini CLI 내부 추론 루프(ReAct 패턴)로 전환 +- anime 하드코딩 핸들러 → MCP 도구로 표준화 (향후 모든 새 기능도 MCP 서버로 추가) +- MCP settings는 홈 레벨 `~/.gemini/settings.json`에 등록 (프로젝트 레벨은 cwd 문제) + +## 변경 파일 + +- `mcp_servers/anime_server.py` — 5개 MCP 도구 (검색/다운/편성표/상태/NAS) +- `mcp_servers/infra_server.py` — 7개 MCP 도구 (Gitea + Vikunja) +- `.gemini/settings.json` — 프로젝트 레벨 MCP 설정 (백업용) +- `prompts/agent.md` — 범용 에이전트 프롬프트 +- `core/gemini_caller.py` — MCP 설정 자동 등록 + 역할별 지시 분기 +- `api/discord_bot.py` — _unified_call→_agent_call, 4-mode 분기 제거, 진행 표시 추가 + +## 미완료 + +- `_handle_anime`, `_handle_task` dead code 아직 남아있음 (실 테스트 후 제거) +- Discord 실행 + 실제 메시지 테스트 미수행 +- Plan→Code→Review 품질 루프 대체 방안 검토 필요 diff --git a/mcp_servers/__init__.py b/mcp_servers/__init__.py new file mode 100644 index 0000000..63cf6f8 --- /dev/null +++ b/mcp_servers/__init__.py @@ -0,0 +1 @@ +# MCP 서버 패키지 diff --git a/mcp_servers/anime_server.py b/mcp_servers/anime_server.py new file mode 100644 index 0000000..c175813 --- /dev/null +++ b/mcp_servers/anime_server.py @@ -0,0 +1,224 @@ +"""MCP 서버 — 애니메이션 도구. + +Gemini CLI에서 MCP로 연결하여 애니 검색/다운로드/편성표 조회를 수행합니다. +stdio 트랜스포트를 사용합니다. +""" + +import asyncio +import sys +from pathlib import Path + +# 프로젝트 루트를 sys.path에 추가 (tools/, config 접근용) +PROJECT_ROOT = Path(__file__).parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +import config # noqa: E402 — .env 로드 + +from mcp.server.fastmcp import FastMCP # noqa: E402 + +mcp = FastMCP("anime") + + +# ────────────────────────────────────────── +# 애니 검색 +# ────────────────────────────────────────── + +@mcp.tool() +async def anime_search(title: str) -> str: + """애니메이션을 검색합니다. + + 제목으로 Anissia에서 애니를 검색하고, 자막 제작자 정보와 + Nyaa 토렌트 검색 결과를 함께 반환합니다. + + Args: + title: 검색할 애니 제목 (한글 또는 일본어) + """ + from tools.anime_pipeline import AnimePipeline + + pipeline = AnimePipeline() + result = await pipeline.search(title) + + if not result.success: + errors = "; ".join(result.errors) if result.errors else "" + return f"검색 실패: {result.message} {errors}" + + anime = result.anime + lines = [ + f"## {anime.subject} ({anime.original_subject})", + f"- 장르: {anime.genres}", + ] + + week_names = ['일', '월', '화', '수', '목', '금', '토', '기타'] + if anime.week is not None: + lines.append(f"- 편성: {week_names[anime.week]}요일 {anime.time}") + lines.append(f"- NAS 폴더: {result.nas_folder}") + + if result.captions: + lines.append(f"\n### 자막 ({len(result.captions)}명)") + for c in result.captions[:5]: + url = f" ({c.website})" if c.website else "" + lines.append(f" - {c.name} — {c.episode}화{url}") + + if result.torrents: + lines.append(f"\n### 토렌트 ({len(result.torrents)}건)") + for t in result.torrents[:5]: + ep = f"{t.episode}화 " if t.episode else "" + lines.append(f" - [{t.group}] {ep}{t.size} (시드: {t.seeders})") + + if result.errors: + lines.append(f"\n### 오류") + for e in result.errors: + lines.append(f" - {e}") + + return "\n".join(lines) + + +# ────────────────────────────────────────── +# 애니 다운로드 +# ────────────────────────────────────────── + +@mcp.tool() +async def anime_download( + title: str, + mode: str = "auto", + episode: int | None = None, +) -> str: + """애니메이션을 다운로드합니다 (자막+영상). + + Anissia에서 검색 후 자막과 토렌트를 NAS에 다운로드합니다. + + Args: + title: 다운로드할 애니 제목 + mode: "auto" (자막+영상), "sub_only" (자막만), "video_only" (영상만) + episode: 특정 에피소드 번호 (None이면 최신) + """ + from tools.anime_pipeline import AnimePipeline + + pipeline = AnimePipeline() + result = await pipeline.download(title, mode=mode, episode=episode) + return result.message + + +# ────────────────────────────────────────── +# 편성표 조회 +# ────────────────────────────────────────── + +@mcp.tool() +async def anime_schedule(weekday: int | None = None, sub_only: bool = False) -> str: + """애니 편성표를 조회합니다. + + Args: + weekday: 요일 번호 (0=일, 1=월, ..., 6=토). None이면 전체 방영 중 목록. + sub_only: True이면 자막 있는 애니만 표시 + """ + from tools.anissia_client import AnissiaClient + + client = AnissiaClient() + + if weekday is not None: + schedule = await client.get_schedule(weekday) + week_names = ['일', '월', '화', '수', '목', '금', '토'] + title = f"{week_names[weekday]}요일 편성표" + else: + schedule = await client.get_all_schedule() + schedule = [a for a in schedule if a.status == "ON"] + title = "현재 방영 중인 애니" + + if sub_only: + schedule = [a for a in schedule if a.caption_count > 0] + title += " (자막 있음)" + + lines = [f"## {title} ({len(schedule)}개)"] + for a in schedule[:30]: + sub = f" 📝{a.caption_count}" if a.caption_count > 0 else "" + lines.append(f"- {a.subject} — {a.time}{sub}") + + if len(schedule) > 30: + lines.append(f"... 외 {len(schedule) - 30}개") + + return "\n".join(lines) + + +# ────────────────────────────────────────── +# 다운로드 상태 +# ────────────────────────────────────────── + +@mcp.tool() +async def anime_download_status() -> str: + """qBittorrent의 현재 다운로드 상태를 확인합니다.""" + from tools.anime_pipeline import AnimePipeline + + pipeline = AnimePipeline() + conn = await pipeline.qbit.test_connection() + + if not conn.get("connected"): + return f"qBittorrent 연결 실패: {conn.get('error', '알 수 없는 오류')}" + + torrents = await pipeline.get_status() + + if not torrents: + return "다운로드 중인 항목이 없습니다." + + lines = [f"## 다운로드 큐 ({len(torrents)}건)"] + for t in torrents[:15]: + icon = "✅" if t["progress"] == "100.0%" else "⏳" + lines.append( + f"- {icon} {t['name'][:50]} — {t['progress']} " + f"({t['speed']}, ETA: {t['eta']})" + ) + + return "\n".join(lines) + + +# ────────────────────────────────────────── +# NAS 애니 목록 +# ────────────────────────────────────────── + +@mcp.tool() +async def anime_nas_list(current_quarter: bool = False, keyword: str = "") -> str: + """NAS에 다운로드된 애니 목록을 조회합니다. + + Args: + current_quarter: True이면 이번 분기 애니만 표시 + keyword: 검색 키워드 (빈 문자열이면 전체) + """ + from tools.nas_scanner import NasScanner + + scanner = NasScanner() + if not scanner.is_accessible(): + return f"NAS 접근 불가: {scanner.base_path}" + + if keyword: + folders = scanner.search(keyword) + elif current_quarter: + folders = scanner.get_current_quarter_anime() + else: + folders = scanner.list_anime_folders() + + if not folders: + return "조건에 맞는 다운로드된 애니가 없습니다." + + total_vids = sum(f.video_count for f in folders) + total_subs = sum(f.subtitle_count for f in folders) + total_size = sum(f.total_size_gb for f in folders) + + lines = [f"## NAS 애니 목록 ({len(folders)}개)"] + for f in folders[:25]: + sub = f" 📝{f.subtitle_count}" if f.subtitle_count > 0 else "" + lines.append( + f"- {f.title} — 🎬{f.video_count}화{sub} ({f.total_size_gb:.1f}GB)" + ) + + if len(folders) > 25: + lines.append(f"... 외 {len(folders) - 25}개") + + lines.append(f"\n총 {total_vids}개 영상 | {total_subs}개 자막 | {total_size:.1f}GB") + return "\n".join(lines) + + +# ────────────────────────────────────────── +# 실행 +# ────────────────────────────────────────── + +if __name__ == "__main__": + mcp.run(transport="stdio") diff --git a/mcp_servers/infra_server.py b/mcp_servers/infra_server.py new file mode 100644 index 0000000..c924693 --- /dev/null +++ b/mcp_servers/infra_server.py @@ -0,0 +1,161 @@ +"""MCP 서버 — 인프라 도구 (Gitea + Vikunja). + +Gemini CLI에서 MCP로 연결하여 Git 저장소 관리, 태스크 관리를 수행합니다. +stdio 트랜스포트를 사용합니다. +""" + +import sys +from pathlib import Path + +# 프로젝트 루트를 sys.path에 추가 +PROJECT_ROOT = Path(__file__).parent.parent +sys.path.insert(0, str(PROJECT_ROOT)) + +import config # noqa: E402 + +from mcp.server.fastmcp import FastMCP # noqa: E402 + +mcp = FastMCP("infra") + + +# ══════════════════════════════════════════ +# Gitea 도구 +# ══════════════════════════════════════════ + +@mcp.tool() +async def gitea_commits(limit: int = 5, branch: str = "main") -> str: + """Gitea 저장소의 최근 커밋 목록을 조회합니다. + + Args: + limit: 조회할 커밋 수 (기본 5) + branch: 브랜치 이름 (기본 main) + """ + from integrations.gitea_client import GiteaClient + + client = GiteaClient() + commits = await client.get_commits(limit=limit, branch=branch) + + if not commits: + return "커밋이 없습니다." + + lines = [f"## 최근 커밋 ({branch}, {len(commits)}개)"] + for c in commits: + lines.append(f"- `{c['sha']}` {c['message']} — {c['author']}") + return "\n".join(lines) + + +@mcp.tool() +async def gitea_prs(state: str = "open") -> str: + """Gitea 저장소의 PR 목록을 조회합니다. + + Args: + state: "open" 또는 "closed" (기본 open) + """ + from integrations.gitea_client import GiteaClient + + client = GiteaClient() + prs = await client.list_prs(state=state) + + if not prs: + return f"{state} 상태의 PR이 없습니다." + + lines = [f"## PR 목록 ({state}, {len(prs)}개)"] + for p in prs: + lines.append(f"- #{p['number']} {p['title']} ({p['state']}, {p['user']})") + return "\n".join(lines) + + +@mcp.tool() +async def gitea_issues(state: str = "open") -> str: + """Gitea 저장소의 이슈 목록을 조회합니다. + + Args: + state: "open" 또는 "closed" (기본 open) + """ + from integrations.gitea_client import GiteaClient + + client = GiteaClient() + issues = await client.list_issues(state=state) + + if not issues: + return f"{state} 상태의 이슈가 없습니다." + + lines = [f"## 이슈 목록 ({state}, {len(issues)}개)"] + for i in issues: + lines.append(f"- #{i['number']} {i['title']} ({i['state']})") + return "\n".join(lines) + + +@mcp.tool() +async def gitea_branches() -> str: + """Gitea 저장소의 브랜치 목록을 조회합니다.""" + from integrations.gitea_client import GiteaClient + + client = GiteaClient() + branches = await client.list_branches() + return "## 브랜치 목록\n" + "\n".join(f"- {b}" for b in branches) + + +# ══════════════════════════════════════════ +# Vikunja 도구 +# ══════════════════════════════════════════ + +@mcp.tool() +async def vikunja_tasks(filter: str = "todo") -> str: + """Vikunja 프로젝트의 태스크 목록을 조회합니다. + + Args: + filter: "todo" (미완료), "done" (완료), "all" (전체) + """ + from integrations.vikunja_client import VikunjaClient + + client = VikunjaClient() + tasks = await client.list_tasks(filter_=filter) + + if not tasks: + return f"{filter} 상태의 태스크가 없습니다." + + lines = [f"## 태스크 ({filter}, {len(tasks)}개)"] + for t in tasks: + icon = "✅" if t["done"] else "⬜" + labels = f" [{', '.join(t['labels'])}]" if t["labels"] else "" + desc = f" — {t['description']}" if t["description"] else "" + lines.append(f"- {icon} #{t['id']} {t['title']}{labels}{desc}") + return "\n".join(lines) + + +@mcp.tool() +async def vikunja_create_task(title: str, description: str = "") -> str: + """Vikunja에 새 태스크를 생성합니다. + + Args: + title: 태스크 제목 + description: 태스크 설명 (선택) + """ + from integrations.vikunja_client import VikunjaClient + + client = VikunjaClient() + result = await client.create_task(title=title, description=description) + return f"태스크 #{result['id']} 생성: {result['title']}" + + +@mcp.tool() +async def vikunja_complete_task(task_id: int) -> str: + """Vikunja 태스크를 완료 처리합니다. + + Args: + task_id: 완료할 태스크 ID + """ + from integrations.vikunja_client import VikunjaClient + + client = VikunjaClient() + result = await client.mark_done(task_id) + return f"태스크 #{task_id} 완료: {result['title']}" + + +# ────────────────────────────────────────── +# 실행 +# ────────────────────────────────────────── + +if __name__ == "__main__": + mcp.run(transport="stdio") diff --git a/prompts/agent.md b/prompts/agent.md new file mode 100644 index 0000000..6994b69 --- /dev/null +++ b/prompts/agent.md @@ -0,0 +1,35 @@ +# Variet Agent + +당신은 **Variet Agent** — 범용 AI 에이전트입니다. +사용자의 요청을 이해하고, 필요한 도구를 자율적으로 선택하여 작업을 완수합니다. + +## 도구 사용 원칙 + +- 도구 없이 답변할 수 있으면 **바로 답변**하세요. +- 도구가 필요하면 호출하고, **결과를 확인한 뒤** 답변하세요. +- 여러 도구를 **순서대로** 사용해야 할 때도 있습니다. +- 도구 호출 결과가 불충분하면 **다른 도구를 시도**하거나 **다른 파라미터**로 재호출하세요. + +## 사용 가능한 도구 영역 + +### 🎬 anime 서버 +- `anime_search` — 애니 검색 (제목, 자막, 토렌트) +- `anime_download` — 애니 다운로드 (자막+영상) +- `anime_schedule` — 편성표 조회 +- `anime_download_status` — qBittorrent 상태 +- `anime_nas_list` — NAS 다운로드 목록 + +### 🔧 infra 서버 +- `gitea_commits`, `gitea_prs`, `gitea_issues`, `gitea_branches` — Git 관리 +- `vikunja_tasks`, `vikunja_create_task`, `vikunja_complete_task` — 태스크 관리 + +### 💻 내장 도구 +- 프로젝트 파일 읽기/쓰기, 쉘 명령 실행 등 Gemini CLI 내장 도구 사용 가능 + +## 응답 규칙 + +- **한국어**로 응답하세요. +- 도구 실행 결과를 사용자에게 **알기 쉽게 정리**하세요. +- 파일 변경 시 **변경 요약**을 제공하세요. +- 에러 발생 시 **원인과 대안**을 안내하세요. +- 불필요하게 길지 않게, **핵심만** 전달하세요.