- Update git remote to new Tailscale hostname - Replace old organization name (hhl-infra) with bandit - Replace old repository name (claude-code-hhl-toolkit) with support-claude-mktplace - Update all documentation references to use generic gitea.example.com - Rebrand from HyperHive Labs to Bandit Labs across all files - Rename workspace file to match new repository name 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
1269 lines
37 KiB
Markdown
1269 lines
37 KiB
Markdown
# Gitea MCP Server Reference
|
|
|
|
## Overview
|
|
|
|
The Gitea MCP Server provides integration with Gitea for issue management, label operations, and repository tracking. It's shared by both `projman` and `projman-pmo` plugins, detecting its operating mode based on environment variables.
|
|
|
|
**Location:** `mcp-servers/gitea/` (repository root)
|
|
|
|
**Key Features:**
|
|
- Issue CRUD operations
|
|
- Label taxonomy management (43-label system)
|
|
- Mode detection (project-scoped vs company-wide)
|
|
- Hybrid configuration (system + project level)
|
|
- Python 3.11+ implementation
|
|
|
|
---
|
|
|
|
## Architecture
|
|
|
|
### Mode Detection
|
|
|
|
The MCP server operates in two modes based on environment variables:
|
|
|
|
**Project Mode (projman):**
|
|
- When `GITEA_REPO` is present
|
|
- Operates on single repository
|
|
- Used by projman plugin
|
|
|
|
**Company Mode (pmo):**
|
|
- When `GITEA_REPO` is absent
|
|
- Operates on all repositories in organization
|
|
- Used by projman-pmo plugin
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/config.py
|
|
def load(self):
|
|
# ... load configs ...
|
|
|
|
self.repo = os.getenv('GITEA_REPO') # Optional
|
|
|
|
if self.repo:
|
|
self.mode = 'project'
|
|
logger.info(f"Running in project mode: {self.repo}")
|
|
else:
|
|
self.mode = 'company'
|
|
logger.info("Running in company-wide mode (PMO)")
|
|
|
|
return {
|
|
'api_url': self.api_url,
|
|
'api_token': self.api_token,
|
|
'owner': self.owner,
|
|
'repo': self.repo,
|
|
'mode': self.mode
|
|
}
|
|
```
|
|
|
|
---
|
|
|
|
## Configuration
|
|
|
|
### System-Level Configuration
|
|
|
|
**File:** `~/.config/claude/gitea.env`
|
|
|
|
```bash
|
|
GITEA_API_URL=https://gitea.example.com/api/v1
|
|
GITEA_API_TOKEN=your_gitea_token
|
|
GITEA_OWNER=bandit
|
|
```
|
|
|
|
**Generating Gitea API Token:**
|
|
|
|
1. Log into Gitea: https://gitea.example.com
|
|
2. Navigate to: **Settings** → **Applications** → **Manage Access Tokens**
|
|
3. Click **Generate New Token**
|
|
4. Token configuration:
|
|
- **Token Name:** `claude-code-mcp`
|
|
- **Required Permissions:**
|
|
- ✅ `repo` (all) - Read/write access to repositories, issues, labels
|
|
- ✅ `read:org` - Read organization information and labels
|
|
- ✅ `read:user` - Read user information
|
|
5. Click **Generate Token**
|
|
6. **Important:** Copy the token immediately (shown only once)
|
|
7. Add to configuration file (see setup below)
|
|
|
|
**Setup:**
|
|
```bash
|
|
# Create config directory
|
|
mkdir -p ~/.config/claude
|
|
|
|
# Create gitea.env
|
|
cat > ~/.config/claude/gitea.env << EOF
|
|
GITEA_API_URL=https://gitea.example.com/api/v1
|
|
GITEA_API_TOKEN=your_token_here
|
|
GITEA_OWNER=bandit
|
|
EOF
|
|
|
|
# Secure the file (important!)
|
|
chmod 600 ~/.config/claude/gitea.env
|
|
|
|
# Verify setup
|
|
cat ~/.config/claude/gitea.env
|
|
```
|
|
|
|
### Project-Level Configuration
|
|
|
|
**File:** `project-root/.env`
|
|
|
|
```bash
|
|
# Repository name (project mode only)
|
|
GITEA_REPO=cuisineflow
|
|
```
|
|
|
|
**Setup:**
|
|
```bash
|
|
# In each project root
|
|
echo "GITEA_REPO=cuisineflow" > .env
|
|
|
|
# Add to .gitignore
|
|
echo ".env" >> .gitignore
|
|
```
|
|
|
|
### Configuration Loading Strategy
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/config.py
|
|
from pathlib import Path
|
|
from dotenv import load_dotenv
|
|
import os
|
|
from typing import Dict, Optional
|
|
|
|
class GiteaConfig:
|
|
"""Hybrid configuration loader"""
|
|
|
|
def __init__(self):
|
|
self.api_url: Optional[str] = None
|
|
self.api_token: Optional[str] = None
|
|
self.owner: Optional[str] = None
|
|
self.repo: Optional[str] = None
|
|
self.mode: str = 'project'
|
|
|
|
def load(self) -> Dict[str, str]:
|
|
"""
|
|
Load configuration from system and project levels.
|
|
Project-level configuration overrides system-level.
|
|
"""
|
|
# Load system config
|
|
system_config = Path.home() / '.config' / 'claude' / 'gitea.env'
|
|
if system_config.exists():
|
|
load_dotenv(system_config)
|
|
else:
|
|
raise FileNotFoundError(
|
|
f"System config not found: {system_config}\n"
|
|
"Create it with: mkdir -p ~/.config/claude && "
|
|
"cat > ~/.config/claude/gitea.env"
|
|
)
|
|
|
|
# Load project config (overrides system)
|
|
project_config = Path.cwd() / '.env'
|
|
if project_config.exists():
|
|
load_dotenv(project_config, override=True)
|
|
|
|
# Extract values
|
|
self.api_url = os.getenv('GITEA_API_URL')
|
|
self.api_token = os.getenv('GITEA_API_TOKEN')
|
|
self.owner = os.getenv('GITEA_OWNER')
|
|
self.repo = os.getenv('GITEA_REPO') # Optional for PMO
|
|
|
|
# Detect mode
|
|
self.mode = 'project' if self.repo else 'company'
|
|
|
|
# Validate required variables
|
|
self._validate()
|
|
|
|
return {
|
|
'api_url': self.api_url,
|
|
'api_token': self.api_token,
|
|
'owner': self.owner,
|
|
'repo': self.repo,
|
|
'mode': self.mode
|
|
}
|
|
|
|
def _validate(self) -> None:
|
|
"""Validate that required configuration is present"""
|
|
required = {
|
|
'GITEA_API_URL': self.api_url,
|
|
'GITEA_API_TOKEN': self.api_token,
|
|
'GITEA_OWNER': self.owner
|
|
}
|
|
|
|
missing = [key for key, value in required.items() if not value]
|
|
|
|
if missing:
|
|
raise ValueError(
|
|
f"Missing required configuration: {', '.join(missing)}\n"
|
|
"Check your ~/.config/claude/gitea.env file"
|
|
)
|
|
```
|
|
|
|
---
|
|
|
|
## Directory Structure
|
|
|
|
```
|
|
mcp-servers/gitea/
|
|
├── .venv/ # Python virtual environment
|
|
├── requirements.txt # Python dependencies
|
|
├── .env.example # Configuration template
|
|
├── mcp_server/
|
|
│ ├── __init__.py
|
|
│ ├── server.py # MCP server entry point
|
|
│ ├── config.py # Configuration loader
|
|
│ ├── gitea_client.py # Gitea API client
|
|
│ └── tools/
|
|
│ ├── __init__.py
|
|
│ ├── issues.py # Issue CRUD tools
|
|
│ └── labels.py # Label management tools
|
|
└── tests/
|
|
├── test_config.py
|
|
├── test_gitea_client.py
|
|
└── test_tools.py
|
|
```
|
|
|
|
---
|
|
|
|
## Dependencies
|
|
|
|
**File:** `mcp-servers/gitea/requirements.txt`
|
|
|
|
```txt
|
|
mcp>=0.9.0 # MCP SDK from Anthropic
|
|
python-dotenv>=1.0.0 # Environment variable loading
|
|
requests>=2.31.0 # HTTP client for Gitea API
|
|
pydantic>=2.5.0 # Data validation
|
|
pytest>=7.4.3 # Testing framework
|
|
pytest-asyncio>=0.23.0 # Async testing support
|
|
```
|
|
|
|
**Python Version:** 3.10+ required
|
|
|
|
**Installation:**
|
|
```bash
|
|
cd mcp-servers/gitea
|
|
python -m venv .venv
|
|
source .venv/bin/activate # Linux/Mac
|
|
# or .venv\Scripts\activate # Windows
|
|
pip install -r requirements.txt
|
|
```
|
|
|
|
---
|
|
|
|
## Gitea API Client
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/gitea_client.py
|
|
import requests
|
|
from typing import List, Dict, Optional
|
|
from .config import GiteaConfig
|
|
|
|
class GiteaClient:
|
|
"""Client for interacting with Gitea API"""
|
|
|
|
def __init__(self):
|
|
config = GiteaConfig()
|
|
config_dict = config.load()
|
|
|
|
self.base_url = config_dict['api_url']
|
|
self.token = config_dict['api_token']
|
|
self.owner = config_dict['owner']
|
|
self.repo = config_dict.get('repo') # Optional for PMO
|
|
self.mode = config_dict['mode']
|
|
|
|
self.session = requests.Session()
|
|
self.session.headers.update({
|
|
'Authorization': f'token {self.token}',
|
|
'Content-Type': 'application/json'
|
|
})
|
|
|
|
def list_issues(
|
|
self,
|
|
state: str = 'open',
|
|
labels: Optional[List[str]] = None,
|
|
repo: Optional[str] = None
|
|
) -> List[Dict]:
|
|
"""
|
|
List issues from Gitea repository.
|
|
|
|
Args:
|
|
state: Issue state (open, closed, all)
|
|
labels: Filter by labels
|
|
repo: Override configured repo (for PMO multi-repo)
|
|
"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues"
|
|
params = {'state': state}
|
|
|
|
if labels:
|
|
params['labels'] = ','.join(labels)
|
|
|
|
response = self.session.get(url, params=params)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_issue(
|
|
self,
|
|
issue_number: int,
|
|
repo: Optional[str] = None
|
|
) -> Dict:
|
|
"""Get specific issue details"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues/{issue_number}"
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def create_issue(
|
|
self,
|
|
title: str,
|
|
body: str,
|
|
labels: Optional[List[str]] = None,
|
|
repo: Optional[str] = None
|
|
) -> Dict:
|
|
"""Create a new issue in Gitea"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues"
|
|
data = {
|
|
'title': title,
|
|
'body': body
|
|
}
|
|
|
|
if labels:
|
|
data['labels'] = labels
|
|
|
|
response = self.session.post(url, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def update_issue(
|
|
self,
|
|
issue_number: int,
|
|
title: Optional[str] = None,
|
|
body: Optional[str] = None,
|
|
state: Optional[str] = None,
|
|
labels: Optional[List[str]] = None,
|
|
repo: Optional[str] = None
|
|
) -> Dict:
|
|
"""Update existing issue"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues/{issue_number}"
|
|
data = {}
|
|
|
|
if title is not None:
|
|
data['title'] = title
|
|
if body is not None:
|
|
data['body'] = body
|
|
if state is not None:
|
|
data['state'] = state
|
|
if labels is not None:
|
|
data['labels'] = labels
|
|
|
|
response = self.session.patch(url, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def add_comment(
|
|
self,
|
|
issue_number: int,
|
|
comment: str,
|
|
repo: Optional[str] = None
|
|
) -> Dict:
|
|
"""Add comment to issue"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/issues/{issue_number}/comments"
|
|
data = {'body': comment}
|
|
|
|
response = self.session.post(url, json=data)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_labels(
|
|
self,
|
|
repo: Optional[str] = None
|
|
) -> List[Dict]:
|
|
"""Get all labels from repository"""
|
|
target_repo = repo or self.repo
|
|
if not target_repo:
|
|
raise ValueError("Repository not specified")
|
|
|
|
url = f"{self.base_url}/repos/{self.owner}/{target_repo}/labels"
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def get_org_labels(self) -> List[Dict]:
|
|
"""Get organization-level labels"""
|
|
url = f"{self.base_url}/orgs/{self.owner}/labels"
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
# PMO-specific methods
|
|
|
|
def list_repos(self) -> List[Dict]:
|
|
"""List all repositories in organization (PMO mode)"""
|
|
url = f"{self.base_url}/orgs/{self.owner}/repos"
|
|
response = self.session.get(url)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
def aggregate_issues(
|
|
self,
|
|
state: str = 'open',
|
|
labels: Optional[List[str]] = None
|
|
) -> Dict[str, List[Dict]]:
|
|
"""
|
|
Fetch issues across all repositories (PMO mode).
|
|
Returns dict keyed by repository name.
|
|
"""
|
|
repos = self.list_repos()
|
|
aggregated = {}
|
|
|
|
for repo in repos:
|
|
repo_name = repo['name']
|
|
try:
|
|
issues = self.list_issues(
|
|
state=state,
|
|
labels=labels,
|
|
repo=repo_name
|
|
)
|
|
if issues:
|
|
aggregated[repo_name] = issues
|
|
except Exception as e:
|
|
# Log error but continue with other repos
|
|
print(f"Error fetching issues from {repo_name}: {e}")
|
|
|
|
return aggregated
|
|
```
|
|
|
|
---
|
|
|
|
## MCP Server Implementation
|
|
|
|
### Overview
|
|
|
|
The MCP (Model Context Protocol) server exposes Gitea tools to Claude Code via JSON-RPC 2.0 over stdio.
|
|
|
|
**Communication Flow:**
|
|
```
|
|
Claude Code <--> stdio <--> MCP Server <--> Gitea API
|
|
```
|
|
|
|
### Server Entry Point
|
|
|
|
**File:** `mcp-servers/gitea/mcp_server/server.py`
|
|
|
|
```python
|
|
"""
|
|
MCP Server entry point for Gitea integration.
|
|
"""
|
|
import asyncio
|
|
import logging
|
|
import json
|
|
from mcp.server import Server
|
|
from mcp.server.stdio import stdio_server
|
|
from mcp.types import Tool, TextContent
|
|
|
|
from .config import GiteaConfig
|
|
from .gitea_client import GiteaClient
|
|
from .tools.issues import IssueTools
|
|
from .tools.labels import LabelTools
|
|
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class GiteaMCPServer:
|
|
"""MCP Server for Gitea integration"""
|
|
|
|
def __init__(self):
|
|
self.server = Server("gitea-mcp")
|
|
self.config = None
|
|
self.client = None
|
|
self.issue_tools = None
|
|
self.label_tools = None
|
|
|
|
async def initialize(self):
|
|
"""Initialize server and load configuration"""
|
|
try:
|
|
config_loader = GiteaConfig()
|
|
self.config = config_loader.load()
|
|
|
|
self.client = GiteaClient()
|
|
self.issue_tools = IssueTools(self.client)
|
|
self.label_tools = LabelTools(self.client)
|
|
|
|
logger.info(f"Gitea MCP Server initialized in {self.config['mode']} mode")
|
|
except Exception as e:
|
|
logger.error(f"Failed to initialize: {e}")
|
|
raise
|
|
|
|
def setup_tools(self):
|
|
"""Register all available tools with the MCP server"""
|
|
|
|
@self.server.list_tools()
|
|
async def list_tools() -> list[Tool]:
|
|
"""Return list of available tools"""
|
|
return [
|
|
Tool(
|
|
name="list_issues",
|
|
description="List issues from Gitea repository",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"state": {
|
|
"type": "string",
|
|
"enum": ["open", "closed", "all"],
|
|
"default": "open"
|
|
},
|
|
"labels": {
|
|
"type": "array",
|
|
"items": {"type": "string"}
|
|
},
|
|
"repo": {"type": "string"}
|
|
}
|
|
}
|
|
),
|
|
Tool(
|
|
name="get_issue",
|
|
description="Get specific issue details",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"issue_number": {"type": "integer"},
|
|
"repo": {"type": "string"}
|
|
},
|
|
"required": ["issue_number"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="create_issue",
|
|
description="Create a new issue in Gitea",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"title": {"type": "string"},
|
|
"body": {"type": "string"},
|
|
"labels": {
|
|
"type": "array",
|
|
"items": {"type": "string"}
|
|
},
|
|
"repo": {"type": "string"}
|
|
},
|
|
"required": ["title", "body"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="update_issue",
|
|
description="Update existing issue",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"issue_number": {"type": "integer"},
|
|
"title": {"type": "string"},
|
|
"body": {"type": "string"},
|
|
"state": {
|
|
"type": "string",
|
|
"enum": ["open", "closed"]
|
|
},
|
|
"labels": {
|
|
"type": "array",
|
|
"items": {"type": "string"}
|
|
},
|
|
"repo": {"type": "string"}
|
|
},
|
|
"required": ["issue_number"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="add_comment",
|
|
description="Add comment to issue",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"issue_number": {"type": "integer"},
|
|
"comment": {"type": "string"},
|
|
"repo": {"type": "string"}
|
|
},
|
|
"required": ["issue_number", "comment"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="get_labels",
|
|
description="Get all available labels (org + repo)",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string"}
|
|
}
|
|
}
|
|
),
|
|
Tool(
|
|
name="suggest_labels",
|
|
description="Analyze context and suggest appropriate labels",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"context": {"type": "string"}
|
|
},
|
|
"required": ["context"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="aggregate_issues",
|
|
description="Fetch issues across all repositories (PMO mode)",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"state": {
|
|
"type": "string",
|
|
"enum": ["open", "closed", "all"],
|
|
"default": "open"
|
|
},
|
|
"labels": {
|
|
"type": "array",
|
|
"items": {"type": "string"}
|
|
}
|
|
}
|
|
}
|
|
)
|
|
]
|
|
|
|
@self.server.call_tool()
|
|
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
|
"""Handle tool invocation"""
|
|
try:
|
|
# Route to appropriate tool handler
|
|
if name == "list_issues":
|
|
result = await self.issue_tools.list_issues(**arguments)
|
|
elif name == "get_issue":
|
|
result = await self.issue_tools.get_issue(**arguments)
|
|
elif name == "create_issue":
|
|
result = await self.issue_tools.create_issue(**arguments)
|
|
elif name == "update_issue":
|
|
result = await self.issue_tools.update_issue(**arguments)
|
|
elif name == "add_comment":
|
|
result = await self.issue_tools.add_comment(**arguments)
|
|
elif name == "get_labels":
|
|
result = await self.label_tools.get_labels(**arguments)
|
|
elif name == "suggest_labels":
|
|
result = await self.label_tools.suggest_labels(**arguments)
|
|
elif name == "aggregate_issues":
|
|
result = await self.issue_tools.aggregate_issues(**arguments)
|
|
else:
|
|
raise ValueError(f"Unknown tool: {name}")
|
|
|
|
return [TextContent(
|
|
type="text",
|
|
text=json.dumps(result, indent=2)
|
|
)]
|
|
|
|
except Exception as e:
|
|
logger.error(f"Tool {name} failed: {e}")
|
|
return [TextContent(
|
|
type="text",
|
|
text=f"Error: {str(e)}"
|
|
)]
|
|
|
|
async def run(self):
|
|
"""Run the MCP server"""
|
|
await self.initialize()
|
|
self.setup_tools()
|
|
|
|
async with stdio_server() as (read_stream, write_stream):
|
|
await self.server.run(
|
|
read_stream,
|
|
write_stream,
|
|
self.server.create_initialization_options()
|
|
)
|
|
|
|
|
|
async def main():
|
|
"""Main entry point"""
|
|
server = GiteaMCPServer()
|
|
await server.run()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|
|
```
|
|
|
|
### Tool Implementation with Async Wrappers
|
|
|
|
Since the Gitea client uses `requests` (synchronous), but MCP tools must be async:
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/tools/issues.py
|
|
import asyncio
|
|
|
|
class IssueTools:
|
|
def __init__(self, gitea_client):
|
|
self.gitea = gitea_client
|
|
|
|
async def list_issues(self, state='open', labels=None, repo=None):
|
|
"""
|
|
List issues from repository.
|
|
|
|
Wraps sync Gitea client method in executor for async compatibility.
|
|
"""
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(
|
|
None,
|
|
self.gitea.list_issues,
|
|
state,
|
|
labels,
|
|
repo
|
|
)
|
|
|
|
async def create_issue(self, title, body, labels=None, repo=None):
|
|
"""Create new issue"""
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(
|
|
None,
|
|
self.gitea.create_issue,
|
|
title,
|
|
body,
|
|
labels,
|
|
repo
|
|
)
|
|
|
|
# Other methods follow same pattern
|
|
```
|
|
|
|
### Branch Detection in MCP Tools
|
|
|
|
Implement branch-aware security at the MCP level:
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/tools/issues.py
|
|
import subprocess
|
|
|
|
class IssueTools:
|
|
def _get_current_branch(self) -> str:
|
|
"""Get current git branch"""
|
|
try:
|
|
result = subprocess.run(
|
|
['git', 'rev-parse', '--abbrev-ref', 'HEAD'],
|
|
capture_output=True,
|
|
text=True,
|
|
check=True
|
|
)
|
|
return result.stdout.strip()
|
|
except subprocess.CalledProcessError:
|
|
return "unknown"
|
|
|
|
def _check_branch_permissions(self, operation: str) -> bool:
|
|
"""Check if operation is allowed on current branch"""
|
|
branch = self._get_current_branch()
|
|
|
|
# Production branches (read-only except incidents)
|
|
if branch in ['main', 'master'] or branch.startswith('prod/'):
|
|
return operation in ['list_issues', 'get_issue', 'get_labels']
|
|
|
|
# Staging branches (read-only for code)
|
|
if branch == 'staging' or branch.startswith('stage/'):
|
|
return operation in ['list_issues', 'get_issue', 'get_labels', 'create_issue']
|
|
|
|
# Development branches (full access)
|
|
if branch in ['development', 'develop'] or branch.startswith(('feat/', 'feature/', 'dev/')):
|
|
return True
|
|
|
|
# Unknown branch - be restrictive
|
|
return False
|
|
|
|
async def create_issue(self, title, body, labels=None, repo=None):
|
|
"""Create issue with branch check"""
|
|
if not self._check_branch_permissions('create_issue'):
|
|
branch = self._get_current_branch()
|
|
raise PermissionError(
|
|
f"Cannot create issues on branch '{branch}'. "
|
|
f"Switch to a development branch to create issues."
|
|
)
|
|
|
|
loop = asyncio.get_event_loop()
|
|
return await loop.run_in_executor(
|
|
None,
|
|
self.gitea.create_issue,
|
|
title,
|
|
body,
|
|
labels,
|
|
repo
|
|
)
|
|
```
|
|
|
|
### Testing the MCP Server
|
|
|
|
**Manual testing:**
|
|
```bash
|
|
cd mcp-servers/gitea
|
|
source .venv/bin/activate
|
|
python -m mcp_server.server
|
|
```
|
|
|
|
**Unit tests with mocks:**
|
|
```python
|
|
# tests/test_server.py
|
|
import pytest
|
|
from unittest.mock import Mock, patch
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_server_initialization():
|
|
"""Test server initializes correctly"""
|
|
from mcp_server.server import GiteaMCPServer
|
|
|
|
with patch('mcp_server.server.GiteaClient'):
|
|
server = GiteaMCPServer()
|
|
await server.initialize()
|
|
|
|
assert server.client is not None
|
|
assert server.config is not None
|
|
```
|
|
|
|
**Integration tests with real API:**
|
|
```python
|
|
# tests/test_integration.py
|
|
import pytest
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.asyncio
|
|
async def test_list_issues_real_api():
|
|
"""Test against real Gitea instance"""
|
|
from mcp_server.server import GiteaMCPServer
|
|
|
|
server = GiteaMCPServer()
|
|
await server.initialize()
|
|
|
|
result = await server.issue_tools.list_issues(state='open')
|
|
assert isinstance(result, list)
|
|
```
|
|
|
|
**Run tests:**
|
|
```bash
|
|
# Unit tests only (with mocks)
|
|
pytest tests/ -m "not integration"
|
|
|
|
# Integration tests (requires Gitea access)
|
|
pytest tests/ -m integration
|
|
|
|
# All tests
|
|
pytest tests/
|
|
```
|
|
|
|
---
|
|
|
|
## Label Taxonomy System
|
|
|
|
### Dynamic Label System
|
|
|
|
The label taxonomy is **fetched dynamically from Gitea** at runtime. The current system has 44 labels:
|
|
|
|
**Organization Labels (28):**
|
|
- Agent/2
|
|
- Complexity/3
|
|
- Efforts/5
|
|
- Priority/4
|
|
- Risk/3
|
|
- Source/4
|
|
- Type/6 (includes Type/Bug, Type/Feature, Type/Refactor, Type/Documentation, Type/Test, Type/Chore)
|
|
|
|
**Repository Labels (16):**
|
|
- Component/9 (Backend, Frontend, API, Database, Auth, Deploy, Testing, Docs, Infra)
|
|
- Tech/7 (Python, JavaScript, Docker, PostgreSQL, Redis, Vue, FastAPI)
|
|
|
|
**Note:** Label count and composition may change. The `/labels-sync` command fetches the latest labels from Gitea and updates suggestion logic accordingly.
|
|
|
|
### Label Suggestion Logic
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/tools/labels.py
|
|
from typing import List, Dict
|
|
import re
|
|
|
|
class LabelTools:
|
|
def __init__(self, gitea_client):
|
|
self.gitea = gitea_client
|
|
|
|
async def get_labels(self, repo: str = None) -> List[Dict]:
|
|
"""Get all labels (org + repo)"""
|
|
org_labels = self.gitea.get_org_labels()
|
|
|
|
if repo or self.gitea.repo:
|
|
target_repo = repo or self.gitea.repo
|
|
repo_labels = self.gitea.get_labels(target_repo)
|
|
return org_labels + repo_labels
|
|
|
|
return org_labels
|
|
|
|
async def suggest_labels(self, context: str) -> List[str]:
|
|
"""
|
|
Analyze context and suggest appropriate labels.
|
|
|
|
Args:
|
|
context: Issue title + description or sprint context
|
|
"""
|
|
suggested = []
|
|
context_lower = context.lower()
|
|
|
|
# Type detection (exclusive - only one)
|
|
if any(word in context_lower for word in ['bug', 'error', 'fix', 'broken']):
|
|
suggested.append('Type/Bug')
|
|
elif any(word in context_lower for word in ['refactor', 'extract', 'restructure', 'architecture', 'service extraction']):
|
|
suggested.append('Type/Refactor')
|
|
elif any(word in context_lower for word in ['feature', 'add', 'implement', 'new']):
|
|
suggested.append('Type/Feature')
|
|
elif any(word in context_lower for word in ['docs', 'documentation', 'readme']):
|
|
suggested.append('Type/Documentation')
|
|
elif any(word in context_lower for word in ['test', 'testing', 'spec']):
|
|
suggested.append('Type/Test')
|
|
elif any(word in context_lower for word in ['chore', 'maintenance', 'update']):
|
|
suggested.append('Type/Chore')
|
|
|
|
# Priority detection
|
|
if any(word in context_lower for word in ['critical', 'urgent', 'blocker', 'blocking']):
|
|
suggested.append('Priority/Critical')
|
|
elif any(word in context_lower for word in ['high', 'important', 'asap']):
|
|
suggested.append('Priority/High')
|
|
elif any(word in context_lower for word in ['low', 'nice-to-have', 'optional']):
|
|
suggested.append('Priority/Low')
|
|
else:
|
|
suggested.append('Priority/Medium')
|
|
|
|
# Component detection (based on keywords)
|
|
component_keywords = {
|
|
'Component/Backend': ['backend', 'server', 'api', 'database', 'service'],
|
|
'Component/Frontend': ['frontend', 'ui', 'interface', 'react', 'vue'],
|
|
'Component/API': ['api', 'endpoint', 'rest', 'graphql'],
|
|
'Component/Database': ['database', 'db', 'sql', 'migration', 'schema'],
|
|
'Component/Auth': ['auth', 'authentication', 'login', 'oauth', 'token'],
|
|
'Component/Deploy': ['deploy', 'deployment', 'docker', 'kubernetes'],
|
|
'Component/Testing': ['test', 'testing', 'spec', 'jest', 'pytest'],
|
|
'Component/Docs': ['docs', 'documentation', 'readme', 'guide']
|
|
}
|
|
|
|
for label, keywords in component_keywords.items():
|
|
if any(keyword in context_lower for keyword in keywords):
|
|
suggested.append(label)
|
|
|
|
# Source detection (based on git branch or context)
|
|
if 'development' in context_lower or 'dev/' in context_lower:
|
|
suggested.append('Source/Development')
|
|
elif 'staging' in context_lower:
|
|
suggested.append('Source/Staging')
|
|
elif 'production' in context_lower or 'prod' in context_lower:
|
|
suggested.append('Source/Production')
|
|
|
|
return suggested
|
|
```
|
|
|
|
---
|
|
|
|
## MCP Tools
|
|
|
|
### Issue Tools
|
|
|
|
```python
|
|
# mcp-servers/gitea/mcp_server/tools/issues.py
|
|
class IssueTools:
|
|
def __init__(self, gitea_client):
|
|
self.gitea = gitea_client
|
|
|
|
async def list_issues(self, state='open', labels=None, repo=None):
|
|
"""List issues in repository"""
|
|
return self.gitea.list_issues(state=state, labels=labels, repo=repo)
|
|
|
|
async def get_issue(self, issue_number, repo=None):
|
|
"""Get specific issue details"""
|
|
return self.gitea.get_issue(issue_number, repo=repo)
|
|
|
|
async def create_issue(self, title, body, labels=None, repo=None):
|
|
"""Create new issue"""
|
|
return self.gitea.create_issue(title, body, labels=labels, repo=repo)
|
|
|
|
async def update_issue(self, issue_number, title=None, body=None, state=None, labels=None, repo=None):
|
|
"""Update existing issue"""
|
|
return self.gitea.update_issue(
|
|
issue_number,
|
|
title=title,
|
|
body=body,
|
|
state=state,
|
|
labels=labels,
|
|
repo=repo
|
|
)
|
|
|
|
async def add_comment(self, issue_number, comment, repo=None):
|
|
"""Add comment to issue"""
|
|
return self.gitea.add_comment(issue_number, comment, repo=repo)
|
|
|
|
# PMO-specific
|
|
async def aggregate_issues(self, state='open', labels=None):
|
|
"""Aggregate issues across all repositories (PMO mode)"""
|
|
if self.gitea.mode != 'company':
|
|
raise ValueError("aggregate_issues only available in company mode")
|
|
return self.gitea.aggregate_issues(state=state, labels=labels)
|
|
```
|
|
|
|
---
|
|
|
|
## Testing
|
|
|
|
### Unit Tests
|
|
|
|
```python
|
|
# tests/test_config.py
|
|
import pytest
|
|
from pathlib import Path
|
|
from mcp_server.config import GiteaConfig
|
|
|
|
def test_load_system_config(tmp_path, monkeypatch):
|
|
"""Test loading system-level configuration"""
|
|
# Mock home directory
|
|
config_dir = tmp_path / '.config' / 'claude'
|
|
config_dir.mkdir(parents=True)
|
|
|
|
config_file = config_dir / 'gitea.env'
|
|
config_file.write_text(
|
|
"GITEA_API_URL=https://test.com/api/v1\n"
|
|
"GITEA_API_TOKEN=test_token\n"
|
|
"GITEA_OWNER=test_owner\n"
|
|
)
|
|
|
|
monkeypatch.setenv('HOME', str(tmp_path))
|
|
|
|
config = GiteaConfig()
|
|
result = config.load()
|
|
|
|
assert result['api_url'] == 'https://test.com/api/v1'
|
|
assert result['api_token'] == 'test_token'
|
|
assert result['owner'] == 'test_owner'
|
|
assert result['mode'] == 'company' # No repo specified
|
|
|
|
def test_project_config_override(tmp_path, monkeypatch):
|
|
"""Test that project config overrides system config"""
|
|
# Set up system config
|
|
system_config_dir = tmp_path / '.config' / 'claude'
|
|
system_config_dir.mkdir(parents=True)
|
|
|
|
system_config = system_config_dir / 'gitea.env'
|
|
system_config.write_text(
|
|
"GITEA_API_URL=https://test.com/api/v1\n"
|
|
"GITEA_API_TOKEN=test_token\n"
|
|
"GITEA_OWNER=test_owner\n"
|
|
)
|
|
|
|
# Set up project config
|
|
project_dir = tmp_path / 'project'
|
|
project_dir.mkdir()
|
|
|
|
project_config = project_dir / '.env'
|
|
project_config.write_text("GITEA_REPO=test_repo\n")
|
|
|
|
monkeypatch.setenv('HOME', str(tmp_path))
|
|
monkeypatch.chdir(project_dir)
|
|
|
|
config = GiteaConfig()
|
|
result = config.load()
|
|
|
|
assert result['repo'] == 'test_repo'
|
|
assert result['mode'] == 'project'
|
|
|
|
def test_missing_required_config():
|
|
"""Test error handling for missing configuration"""
|
|
with pytest.raises(FileNotFoundError):
|
|
config = GiteaConfig()
|
|
config.load()
|
|
```
|
|
|
|
### Integration Tests
|
|
|
|
```python
|
|
# tests/test_gitea_client.py
|
|
import pytest
|
|
from mcp_server.gitea_client import GiteaClient
|
|
|
|
@pytest.fixture
|
|
def gitea_client():
|
|
"""Fixture providing configured Gitea client"""
|
|
return GiteaClient()
|
|
|
|
def test_list_issues(gitea_client):
|
|
"""Test listing issues from Gitea"""
|
|
issues = gitea_client.list_issues(state='open')
|
|
assert isinstance(issues, list)
|
|
|
|
def test_create_issue(gitea_client):
|
|
"""Test creating an issue in Gitea"""
|
|
issue = gitea_client.create_issue(
|
|
title="Test Issue",
|
|
body="Test body",
|
|
labels=["Type/Bug", "Priority/Low"]
|
|
)
|
|
assert issue['title'] == "Test Issue"
|
|
assert any(label['name'] == "Type/Bug" for label in issue['labels'])
|
|
|
|
def test_get_labels(gitea_client):
|
|
"""Test fetching labels"""
|
|
labels = gitea_client.get_labels()
|
|
assert isinstance(labels, list)
|
|
assert len(labels) > 0
|
|
|
|
def test_pmo_mode_aggregate_issues():
|
|
"""Test PMO mode aggregation (no repo specified)"""
|
|
# Set up client without repo
|
|
client = GiteaClient() # Should detect company mode
|
|
|
|
if client.mode == 'company':
|
|
aggregated = client.aggregate_issues(state='open')
|
|
assert isinstance(aggregated, dict)
|
|
# Each key should be a repo name
|
|
for repo_name, issues in aggregated.items():
|
|
assert isinstance(issues, list)
|
|
```
|
|
|
|
### Running Tests
|
|
|
|
```bash
|
|
# Activate virtual environment
|
|
source .venv/bin/activate
|
|
|
|
# Run all tests
|
|
pytest
|
|
|
|
# Run with coverage
|
|
pytest --cov=mcp_server --cov-report=html
|
|
|
|
# Run specific test file
|
|
pytest tests/test_config.py
|
|
|
|
# Run with verbose output
|
|
pytest -v
|
|
```
|
|
|
|
---
|
|
|
|
## Performance Optimization
|
|
|
|
### Caching
|
|
|
|
```python
|
|
from functools import lru_cache
|
|
|
|
@lru_cache(maxsize=128)
|
|
def get_labels_cached(self, repo: str) -> List[Dict]:
|
|
"""Cached label retrieval to reduce API calls"""
|
|
return self.get_labels(repo)
|
|
```
|
|
|
|
### Retry Logic
|
|
|
|
```python
|
|
import time
|
|
from typing import Callable
|
|
|
|
def retry_on_failure(max_retries=3, delay=1):
|
|
"""Decorator for retrying failed API calls"""
|
|
def decorator(func: Callable):
|
|
def wrapper(*args, **kwargs):
|
|
for attempt in range(max_retries):
|
|
try:
|
|
return func(*args, **kwargs)
|
|
except requests.exceptions.RequestException as e:
|
|
if attempt == max_retries - 1:
|
|
raise
|
|
time.sleep(delay * (attempt + 1))
|
|
return wrapper
|
|
return decorator
|
|
|
|
@retry_on_failure(max_retries=3)
|
|
def list_issues(self, state='open', labels=None, repo=None):
|
|
"""List issues with automatic retry"""
|
|
# Implementation
|
|
```
|
|
|
|
---
|
|
|
|
## Troubleshooting
|
|
|
|
### Common Issues
|
|
|
|
**Issue:** Module not found
|
|
```bash
|
|
# Solution: Ensure PYTHONPATH is set in .mcp.json
|
|
"env": {
|
|
"PYTHONPATH": "${CLAUDE_PLUGIN_ROOT}/../mcp-servers/gitea"
|
|
}
|
|
```
|
|
|
|
**Issue:** Configuration not loading
|
|
```bash
|
|
# Solution: Check file permissions
|
|
chmod 600 ~/.config/claude/gitea.env
|
|
|
|
# Verify file exists
|
|
cat ~/.config/claude/gitea.env
|
|
```
|
|
|
|
**Issue:** API authentication failing
|
|
```bash
|
|
# Solution: Test token manually
|
|
curl -H "Authorization: token YOUR_TOKEN" \
|
|
https://your-gitea.com/api/v1/user
|
|
```
|
|
|
|
**Issue:** PMO mode not working
|
|
```bash
|
|
# Solution: Ensure GITEA_REPO is NOT set
|
|
# Check environment variables
|
|
env | grep GITEA
|
|
```
|
|
|
|
---
|
|
|
|
## Security
|
|
|
|
### Best Practices
|
|
|
|
1. **Token Storage:**
|
|
- Store tokens in `~/.config/claude/gitea.env`
|
|
- Set file permissions to 600
|
|
- Never commit tokens to git
|
|
|
|
2. **Input Validation:**
|
|
- Validate all user input before API calls
|
|
- Sanitize issue titles and descriptions
|
|
- Prevent injection attacks
|
|
|
|
3. **Error Handling:**
|
|
- Don't leak tokens in error messages
|
|
- Log errors without sensitive data
|
|
- Provide user-friendly error messages
|
|
|
|
4. **API Rate Limiting:**
|
|
- Implement exponential backoff
|
|
- Cache frequently accessed data
|
|
- Batch requests where possible
|
|
|
|
---
|
|
|
|
## Next Steps
|
|
|
|
1. **Set up system configuration**
|
|
2. **Create virtual environment**
|
|
3. **Install dependencies**
|
|
4. **Run tests against actual Gitea instance**
|
|
5. **Integrate with projman plugin**
|