Initial commit: Gitea MCP Server

- 6 curated MCP tools (get_my_user_info, search_repos, list_my_repos, get_repo, list_repo_issues, list_repo_commits)
- API pass-through tool (gitea_api_call) for complete API coverage
- Curated API reference resource (gitea://api-reference)
- Health check endpoint
- Docker support with health checks
This commit is contained in:
Ben
2025-12-22 04:45:30 +00:00
commit 10906ab115
8 changed files with 22356 additions and 0 deletions

2
.env.example Normal file
View File

@@ -0,0 +1,2 @@
GITEA_URL=https://gitea.example.com
GITEA_TOKEN=your_access_token

26
.gitignore vendored Normal file
View File

@@ -0,0 +1,26 @@
.env
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
*.egg-info/
.installed.cfg
*.egg
.venv/
venv/
ENV/
.pytest_cache/
.mypy_cache/

24
Dockerfile Normal file
View File

@@ -0,0 +1,24 @@
FROM python:3.12-slim
WORKDIR /app
# Install uv for fast dependency management
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
# Copy project files
COPY pyproject.toml .
COPY server.py .
# Install dependencies
RUN uv pip install --system --no-cache .
# Install curl for health checks
RUN apt-get update && apt-get install -y --no-install-recommends curl && \
rm -rf /var/lib/apt/lists/*
EXPOSE 8000
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
CMD ["python", "server.py"]

92
README.md Normal file
View File

@@ -0,0 +1,92 @@
# Gitea MCP Server
A lightweight Model Context Protocol (MCP) server for Gitea, following the **Hybrid MCP Light** pattern.
## Features
- **5 Curated Tools** for common operations
- **API Pass-through** for complete API coverage
- **Embedded API Reference** for agent self-service
- **Health Check Endpoint** for Docker/Kubernetes
## Tools
| Tool | Description |
|:-----|:------------|
| `get_my_user_info` | Get authenticated user info |
| `search_repos` | Search repositories by keyword |
| `list_my_repos` | List user's accessible repositories |
| `get_repo` | Get repository details |
| `list_repo_issues` | List issues for a repository |
| `list_repo_commits` | List commits for a repository |
| `gitea_api_call` | Raw API pass-through for any endpoint |
## Resources
| Resource URI | Description |
|:-------------|:------------|
| `gitea://api-reference` | Quick reference for common API endpoints |
## Setup
### Environment Variables
```bash
GITEA_URL=https://gitea.example.com
GITEA_TOKEN=your_access_token
```
### Local Development
```bash
# Install dependencies
uv pip install -e .
# Run server
python server.py
```
### Docker
```bash
# Build and run
docker compose up -d
# Check health
curl http://localhost:8000/health
```
## MCP Client Configuration
### Gemini CLI (`~/.gemini/settings.json`)
```json
{
"mcpServers": {
"gitea": {
"url": "http://localhost:8000/mcp"
}
}
}
```
## API Pass-through Examples
```python
# Create a release
gitea_api_call(
endpoint="/repos/myorg/myrepo/releases",
method="POST",
body='{"tag_name": "v1.0.0", "name": "Release 1.0"}'
)
# Get file contents
gitea_api_call(
endpoint="/repos/myorg/myrepo/contents/README.md",
params='{"ref": "main"}'
)
```
## License
MIT

15
docker-compose.yml Normal file
View File

@@ -0,0 +1,15 @@
services:
gitea-mcp:
build: .
container_name: gitea-mcp
restart: unless-stopped
ports:
- "8000:8000"
env_file:
- .env
healthcheck:
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
interval: 30s
timeout: 10s
retries: 3
start_period: 5s

21796
plugin-redoc-2.yaml Normal file

File diff suppressed because it is too large Load Diff

27
pyproject.toml Normal file
View File

@@ -0,0 +1,27 @@
[project]
name = "gitea-mcp"
version = "0.1.0"
description = "MCP server for Gitea API"
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"fastmcp>=2.0",
"httpx>=0.27",
"starlette>=0.40",
"uvicorn>=0.30",
"python-dotenv>=1.0",
]
[project.optional-dependencies]
dev = [
"pytest>=8.0",
"pytest-asyncio>=0.24",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[tool.hatch.build.targets.wheel]
packages = ["."]
include = ["server.py"]

374
server.py Normal file
View File

@@ -0,0 +1,374 @@
"""
Gitea MCP Server - Hybrid MCP Light implementation for Gitea API.
Provides 5 curated tools plus an API pass-through for complete API coverage.
"""
import json
import os
from contextlib import asynccontextmanager
from typing import Any
from urllib.parse import urljoin
import httpx
from dotenv import load_dotenv
from fastmcp import FastMCP
from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Mount, Route
load_dotenv()
# Configuration
GITEA_URL = os.getenv("GITEA_URL", "").rstrip("/")
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
# Initialize MCP server
mcp = FastMCP(
"Gitea MCP",
instructions="MCP server for Gitea API - provides repository, issue, and user management",
)
class GiteaClient:
"""HTTP client for Gitea API with token authentication."""
def __init__(self, base_url: str, token: str):
self.base_url = base_url.rstrip("/")
self.api_url = f"{self.base_url}/api/v1"
self.token = token
self._client: httpx.AsyncClient | None = None
async def _get_client(self) -> httpx.AsyncClient:
if self._client is None:
self._client = httpx.AsyncClient(
headers={
"Authorization": f"token {self.token}",
"Accept": "application/json",
"Content-Type": "application/json",
},
timeout=30.0,
)
return self._client
async def close(self):
if self._client:
await self._client.aclose()
self._client = None
async def request(
self,
method: str,
endpoint: str,
params: dict[str, Any] | None = None,
json_body: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""Execute an API request to Gitea."""
client = await self._get_client()
url = f"{self.api_url}{endpoint}"
try:
response = await client.request(
method=method.upper(),
url=url,
params=params,
json=json_body,
)
response.raise_for_status()
if response.status_code == 204:
return {"status": "success", "message": "No content"}
return response.json()
except httpx.HTTPStatusError as e:
return {
"error": True,
"status_code": e.response.status_code,
"message": str(e),
"detail": e.response.text[:500] if e.response.text else None,
}
except Exception as e:
return {"error": True, "message": str(e)}
async def health_check(self) -> bool:
"""Check if Gitea is accessible."""
try:
client = await self._get_client()
response = await client.get(f"{self.api_url}/version")
return response.status_code == 200
except Exception:
return False
# Global client instance
client = GiteaClient(GITEA_URL, GITEA_TOKEN)
# =============================================================================
# Curated API Reference
# =============================================================================
API_REFERENCE = """
# Gitea API Quick Reference
For complete API documentation, see: {base_url}/api/swagger
## Common Endpoints for `gitea_api_call` tool
### Repository Operations
- GET `/repos/search?q={{keyword}}&limit=10` - Search repositories
- GET `/repos/{{owner}}/{{repo}}` - Get repository details
- POST `/repos/{{owner}}/{{repo}}` - Create repository (body: name, description, private)
- PATCH `/repos/{{owner}}/{{repo}}` - Update repository
- DELETE `/repos/{{owner}}/{{repo}}` - Delete repository
- GET `/repos/{{owner}}/{{repo}}/branches` - List branches
- GET `/repos/{{owner}}/{{repo}}/commits?limit=20` - List commits
### Issues
- GET `/repos/{{owner}}/{{repo}}/issues?state=open` - List issues
- POST `/repos/{{owner}}/{{repo}}/issues` - Create issue (body: title, body)
- GET `/repos/{{owner}}/{{repo}}/issues/{{index}}` - Get issue by number
- PATCH `/repos/{{owner}}/{{repo}}/issues/{{index}}` - Update issue
- POST `/repos/{{owner}}/{{repo}}/issues/{{index}}/comments` - Add comment (body: body)
### Pull Requests
- GET `/repos/{{owner}}/{{repo}}/pulls?state=open` - List PRs
- POST `/repos/{{owner}}/{{repo}}/pulls` - Create PR (body: title, head, base)
- GET `/repos/{{owner}}/{{repo}}/pulls/{{index}}` - Get PR by number
- POST `/repos/{{owner}}/{{repo}}/pulls/{{index}}/merge` - Merge PR
### File Operations
- GET `/repos/{{owner}}/{{repo}}/contents/{{filepath}}?ref={{branch}}` - Get file content
- PUT `/repos/{{owner}}/{{repo}}/contents/{{filepath}}` - Create/update file
- DELETE `/repos/{{owner}}/{{repo}}/contents/{{filepath}}` - Delete file
### Releases & Tags
- GET `/repos/{{owner}}/{{repo}}/releases` - List releases
- GET `/repos/{{owner}}/{{repo}}/releases/latest` - Get latest release
- POST `/repos/{{owner}}/{{repo}}/releases` - Create release
- GET `/repos/{{owner}}/{{repo}}/tags` - List tags
- POST `/repos/{{owner}}/{{repo}}/tags` - Create tag
### Actions/CI (Gitea Actions)
- GET `/repos/{{owner}}/{{repo}}/actions/runs` - List workflow runs
- GET `/repos/{{owner}}/{{repo}}/actions/jobs?status={{status}}` - List jobs
### User & Organizations
- GET `/user` - Get authenticated user
- GET `/users/{{username}}` - Get user by username
- GET `/user/repos?limit=50` - List authenticated user's repos
- GET `/orgs/{{org}}` - Get organization
- GET `/orgs/{{org}}/repos` - List organization repos
- GET `/orgs/{{org}}/members` - List organization members
### Pagination
Most list endpoints support `page` and `limit` query parameters.
Default limit is usually 20-50 items.
""".format(base_url=GITEA_URL)
@mcp.resource("gitea://api-reference")
def get_api_reference() -> str:
"""Returns the Gitea API quick reference for using the gitea_api_call tool."""
return API_REFERENCE
# =============================================================================
# MCP Tools - Curated Operations
# =============================================================================
@mcp.tool()
async def get_my_user_info() -> str:
"""Get information about the authenticated user.
Returns the current user's profile including username, email, and permissions.
This is useful for determining the authenticated identity before other operations.
"""
result = await client.request("GET", "/user")
return json.dumps(result)
@mcp.tool()
async def search_repos(
keyword: str,
limit: int = 20,
private: bool | None = None,
archived: bool | None = None,
) -> str:
"""Search for repositories by keyword.
Args:
keyword: Search term to find in repository names/descriptions
limit: Maximum number of results (default: 20, max: 50)
private: Filter by private status (optional)
archived: Filter by archived status (optional)
"""
params: dict[str, Any] = {"q": keyword, "limit": min(limit, 50)}
if private is not None:
params["private"] = private
if archived is not None:
params["archived"] = archived
result = await client.request("GET", "/repos/search", params=params)
return json.dumps(result)
@mcp.tool()
async def list_my_repos(
limit: int = 50,
page: int = 1,
) -> str:
"""List repositories owned by or accessible to the authenticated user.
Args:
limit: Maximum number of results per page (default: 50)
page: Page number for pagination (default: 1)
"""
params = {"limit": limit, "page": page}
result = await client.request("GET", "/user/repos", params=params)
return json.dumps(result)
@mcp.tool()
async def get_repo(owner: str, repo: str) -> str:
"""Get detailed information about a specific repository.
Args:
owner: Repository owner (username or organization)
repo: Repository name
"""
result = await client.request("GET", f"/repos/{owner}/{repo}")
return json.dumps(result)
@mcp.tool()
async def list_repo_issues(
owner: str,
repo: str,
state: str = "open",
limit: int = 30,
page: int = 1,
) -> str:
"""List issues for a repository.
Args:
owner: Repository owner (username or organization)
repo: Repository name
state: Issue state filter: 'open', 'closed', or 'all' (default: 'open')
limit: Maximum number of results (default: 30)
page: Page number for pagination (default: 1)
"""
params = {"state": state, "limit": limit, "page": page}
result = await client.request("GET", f"/repos/{owner}/{repo}/issues", params=params)
return json.dumps(result)
@mcp.tool()
async def list_repo_commits(
owner: str,
repo: str,
sha: str | None = None,
limit: int = 30,
page: int = 1,
) -> str:
"""List commits for a repository.
Args:
owner: Repository owner (username or organization)
repo: Repository name
sha: SHA or branch to start listing from (default: default branch)
limit: Maximum number of results (default: 30)
page: Page number for pagination (default: 1)
"""
params: dict[str, Any] = {"limit": limit, "page": page}
if sha:
params["sha"] = sha
result = await client.request("GET", f"/repos/{owner}/{repo}/commits", params=params)
return json.dumps(result)
# =============================================================================
# API Pass-through Tool
# =============================================================================
@mcp.tool()
async def gitea_api_call(
endpoint: str,
method: str = "GET",
params: str = "{}",
body: str = "{}",
) -> str:
"""Execute a raw API call to Gitea.
Use this for any operation not covered by the other tools.
Refer to the 'gitea://api-reference' resource for common endpoints,
or see the full API docs at {base_url}/api/swagger
Args:
endpoint: API endpoint path (e.g., '/repos/owner/repo/releases')
method: HTTP method (GET, POST, PUT, PATCH, DELETE)
params: JSON string of query parameters (optional)
body: JSON string of request body for POST/PUT/PATCH (optional)
Example:
gitea_api_call('/repos/myorg/myrepo/releases', 'POST',
body='{{"tag_name": "v1.0.0", "name": "Release 1.0"}}')
""".format(base_url=GITEA_URL)
try:
params_dict = json.loads(params) if params else {}
body_dict = json.loads(body) if body else {}
except json.JSONDecodeError as e:
return json.dumps({"error": True, "message": f"Invalid JSON: {e}"})
result = await client.request(
method=method,
endpoint=endpoint,
params=params_dict if params_dict else None,
json_body=body_dict if body_dict else None,
)
return json.dumps(result)
# =============================================================================
# Starlette Wrapper for Health Checks
# =============================================================================
async def health_check(request):
"""Health check endpoint for Docker/Kubernetes."""
is_healthy = await client.health_check()
if is_healthy:
return JSONResponse({"status": "ok", "gitea_url": GITEA_URL})
return JSONResponse(
{"status": "unhealthy", "message": "Cannot connect to Gitea"},
status_code=503,
)
@asynccontextmanager
async def lifespan(app):
"""Manage client lifecycle."""
yield
await client.close()
def create_app() -> Starlette:
"""Create the Starlette application with health check and MCP."""
mcp_app = mcp.http_app()
# Add health check route directly to the MCP app
mcp_app.add_route("/health", health_check, methods=["GET"])
return mcp_app
app = create_app()
if __name__ == "__main__":
import uvicorn
port = int(os.getenv("PORT", "8000"))
uvicorn.run(app, host="0.0.0.0", port=port)