Files
leo-claude-mktplace/docs/references/MCP-GITEA.md
lmiranda ba599e342e refactor: update repository URL and rebrand to Bandit Labs
- 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>
2025-12-09 11:51:13 -05:00

37 KiB

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
# 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

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: SettingsApplicationsManage 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:

# 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

# Repository name (project mode only)
GITEA_REPO=cuisineflow

Setup:

# In each project root
echo "GITEA_REPO=cuisineflow" > .env

# Add to .gitignore
echo ".env" >> .gitignore

Configuration Loading Strategy

# 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

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:

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

# 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

"""
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:

# 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:

# 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:

cd mcp-servers/gitea
source .venv/bin/activate
python -m mcp_server.server

Unit tests with mocks:

# 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:

# 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:

# 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

# 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

# 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

# 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

# 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

# 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

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

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

# Solution: Ensure PYTHONPATH is set in .mcp.json
"env": {
  "PYTHONPATH": "${CLAUDE_PLUGIN_ROOT}/../mcp-servers/gitea"
}

Issue: Configuration not loading

# Solution: Check file permissions
chmod 600 ~/.config/claude/gitea.env

# Verify file exists
cat ~/.config/claude/gitea.env

Issue: API authentication failing

# Solution: Test token manually
curl -H "Authorization: token YOUR_TOKEN" \
  https://your-gitea.com/api/v1/user

Issue: PMO mode not working

# 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