162 lines
5.0 KiB
Python
162 lines
5.0 KiB
Python
"""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")
|