generated from personal-projects/leo-claude-mktplace
Compare commits
18 Commits
feat/3-iss
...
608b488763
| Author | SHA1 | Date | |
|---|---|---|---|
| 608b488763 | |||
| 49f2d0bdbb | |||
| f2cba079eb | |||
| 4e81b9bb96 | |||
| d21f85545b | |||
| 3d1fd2e2a6 | |||
| 2fc43ff5c3 | |||
| d11649071e | |||
| 42d625c27f | |||
| 6beb8026df | |||
| acacefeaed | |||
| 604661f096 | |||
| b94dcebfc7 | |||
| 201cc680ca | |||
| 0653a4f70e | |||
| 13ffd8a543 | |||
| 2230bceb51 | |||
| ab8c9069da |
67
.dockerignore
Normal file
67
.dockerignore
Normal file
@@ -0,0 +1,67 @@
|
|||||||
|
# Python
|
||||||
|
__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
|
||||||
|
MANIFEST
|
||||||
|
*.coverage
|
||||||
|
.pytest_cache/
|
||||||
|
.mypy_cache/
|
||||||
|
.ruff_cache/
|
||||||
|
|
||||||
|
# Virtual environments
|
||||||
|
venv/
|
||||||
|
env/
|
||||||
|
ENV/
|
||||||
|
.venv/
|
||||||
|
|
||||||
|
# IDE
|
||||||
|
.vscode/
|
||||||
|
.idea/
|
||||||
|
*.swp
|
||||||
|
*.swo
|
||||||
|
*~
|
||||||
|
|
||||||
|
# Git
|
||||||
|
.git/
|
||||||
|
.gitignore
|
||||||
|
.gitattributes
|
||||||
|
|
||||||
|
# Documentation
|
||||||
|
*.md
|
||||||
|
docs/
|
||||||
|
|
||||||
|
# CI/CD
|
||||||
|
.github/
|
||||||
|
.gitlab-ci.yml
|
||||||
|
|
||||||
|
# Environment files
|
||||||
|
.env
|
||||||
|
.env.*
|
||||||
|
!.env.example
|
||||||
|
|
||||||
|
# Claude Code
|
||||||
|
.claude/
|
||||||
|
|
||||||
|
# Docker
|
||||||
|
.dockerignore
|
||||||
|
docker-compose.override.yml
|
||||||
|
|
||||||
|
# Test files
|
||||||
|
tests/
|
||||||
19
.env.docker.example
Normal file
19
.env.docker.example
Normal file
@@ -0,0 +1,19 @@
|
|||||||
|
# Docker Compose Environment Variables
|
||||||
|
# Copy this file to .env and fill in your values
|
||||||
|
|
||||||
|
# Gitea Configuration (REQUIRED)
|
||||||
|
GITEA_URL=https://gitea.example.com
|
||||||
|
GITEA_TOKEN=your_gitea_api_token_here
|
||||||
|
GITEA_OWNER=your_username_or_org
|
||||||
|
GITEA_REPO=your_repo_name
|
||||||
|
|
||||||
|
# Authentication Configuration (OPTIONAL)
|
||||||
|
# Uncomment to enable Bearer token authentication
|
||||||
|
# AUTH_TOKEN=your_bearer_token_here
|
||||||
|
|
||||||
|
# Tool Filtering Configuration (OPTIONAL)
|
||||||
|
# Uncomment to enable specific tools only (whitelist mode)
|
||||||
|
# ENABLED_TOOLS=list_issues,create_issue,update_issue,list_labels
|
||||||
|
|
||||||
|
# Uncomment to disable specific tools (blacklist mode)
|
||||||
|
# DISABLED_TOOLS=delete_issue,close_milestone
|
||||||
16
.env.example
Normal file
16
.env.example
Normal file
@@ -0,0 +1,16 @@
|
|||||||
|
# Gitea Configuration
|
||||||
|
GITEA_URL=https://gitea.example.com
|
||||||
|
GITEA_TOKEN=your_gitea_api_token_here
|
||||||
|
GITEA_OWNER=your_username_or_org
|
||||||
|
GITEA_REPO=your_repo_name
|
||||||
|
|
||||||
|
# HTTP Server Configuration
|
||||||
|
HTTP_HOST=127.0.0.1
|
||||||
|
HTTP_PORT=8000
|
||||||
|
|
||||||
|
# Authentication Configuration (Optional)
|
||||||
|
# AUTH_TOKEN=your_bearer_token_here
|
||||||
|
|
||||||
|
# Tool Filtering Configuration (Optional)
|
||||||
|
# ENABLED_TOOLS=list_issues,create_issue,update_issue
|
||||||
|
# DISABLED_TOOLS=delete_issue,close_milestone
|
||||||
43
CHANGELOG.md
Normal file
43
CHANGELOG.md
Normal file
@@ -0,0 +1,43 @@
|
|||||||
|
# Changelog
|
||||||
|
|
||||||
|
All notable changes to this project will be documented in this file.
|
||||||
|
|
||||||
|
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
|
||||||
|
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||||
|
|
||||||
|
## [1.0.0] - 2025-02-03
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
- **BREAKING**: Complete architectural rebuild from standalone MCP server to HTTP wrapper pattern
|
||||||
|
- **BREAKING**: Now wraps official `gitea-mcp-server` package instead of implementing Gitea operations directly
|
||||||
|
- Project renamed from standalone implementation to HTTP transport wrapper
|
||||||
|
|
||||||
|
### Added
|
||||||
|
- HTTP transport layer via Starlette/uvicorn for Claude Desktop compatibility
|
||||||
|
- Configuration management module (`config/`) with environment variable support
|
||||||
|
- Tool filtering module (`filtering/`) for Claude Desktop compatibility controls
|
||||||
|
- Bearer token authentication middleware (`middleware/auth.py`)
|
||||||
|
- Comprehensive test suite (30 tests covering all modules)
|
||||||
|
- Docker deployment infrastructure with docker-compose.yml
|
||||||
|
- Health check endpoints (`/health`, `/healthz`, `/ping`)
|
||||||
|
- Deployment documentation and Docker guides
|
||||||
|
- Environment variable configuration with `.env` support
|
||||||
|
|
||||||
|
### Removed
|
||||||
|
- Standalone MCP tool implementations (now delegated to wrapped `gitea-mcp-server`)
|
||||||
|
- Direct Gitea API integration code (handled by wrapped server)
|
||||||
|
|
||||||
|
## [0.1.0] - 2025-01-XX (Initial Standalone Implementation)
|
||||||
|
|
||||||
|
### Added
|
||||||
|
- Initial Python project structure
|
||||||
|
- MCP server core with stdio transport
|
||||||
|
- Issue operations (create, update, list, get)
|
||||||
|
- Label operations (add, remove, list)
|
||||||
|
- Milestone operations (create, update, list)
|
||||||
|
- Authentication with Gitea API tokens
|
||||||
|
- Comprehensive README and documentation
|
||||||
|
|
||||||
|
### Notes
|
||||||
|
- This version was a standalone MCP server implementation
|
||||||
|
- Superseded by HTTP wrapper architecture in Sprint 02
|
||||||
712
DEPLOYMENT.md
Normal file
712
DEPLOYMENT.md
Normal file
@@ -0,0 +1,712 @@
|
|||||||
|
# Deployment Guide
|
||||||
|
|
||||||
|
This guide covers production deployment of the Gitea HTTP MCP Wrapper in various environments.
|
||||||
|
|
||||||
|
## Table of Contents
|
||||||
|
|
||||||
|
1. [Prerequisites](#prerequisites)
|
||||||
|
2. [Docker Deployment](#docker-deployment)
|
||||||
|
3. [Security Best Practices](#security-best-practices)
|
||||||
|
4. [Monitoring and Health Checks](#monitoring-and-health-checks)
|
||||||
|
5. [Reverse Proxy Configuration](#reverse-proxy-configuration)
|
||||||
|
6. [Cloud Deployment](#cloud-deployment)
|
||||||
|
7. [Kubernetes Deployment](#kubernetes-deployment)
|
||||||
|
8. [Troubleshooting](#troubleshooting)
|
||||||
|
|
||||||
|
## Prerequisites
|
||||||
|
|
||||||
|
### Required
|
||||||
|
|
||||||
|
- Docker and Docker Compose (for Docker deployment)
|
||||||
|
- Gitea instance with API access
|
||||||
|
- Gitea API token with appropriate permissions
|
||||||
|
- Network connectivity between wrapper and Gitea instance
|
||||||
|
|
||||||
|
### Recommended
|
||||||
|
|
||||||
|
- HTTPS-capable reverse proxy (Nginx, Caddy, Traefik)
|
||||||
|
- Secrets management solution (not `.env` files in production)
|
||||||
|
- Monitoring and logging infrastructure
|
||||||
|
- Firewall or VPN for network security
|
||||||
|
|
||||||
|
## Docker Deployment
|
||||||
|
|
||||||
|
### Quick Start
|
||||||
|
|
||||||
|
1. **Clone the repository:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git clone https://github.com/lmiranda/gitea-mcp-remote.git
|
||||||
|
cd gitea-mcp-remote
|
||||||
|
```
|
||||||
|
|
||||||
|
2. **Create configuration:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
cp .env.docker.example .env
|
||||||
|
nano .env # Edit with your values
|
||||||
|
```
|
||||||
|
|
||||||
|
Required configuration:
|
||||||
|
```bash
|
||||||
|
GITEA_URL=https://gitea.example.com
|
||||||
|
GITEA_TOKEN=your_gitea_api_token
|
||||||
|
GITEA_OWNER=your_username_or_org
|
||||||
|
GITEA_REPO=your_default_repo
|
||||||
|
AUTH_TOKEN=your_bearer_token # Recommended
|
||||||
|
```
|
||||||
|
|
||||||
|
3. **Start the service:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker-compose up -d
|
||||||
|
```
|
||||||
|
|
||||||
|
4. **Verify deployment:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
curl http://localhost:8000/health
|
||||||
|
```
|
||||||
|
|
||||||
|
### Production Configuration
|
||||||
|
|
||||||
|
For production, use a more robust `docker-compose.yml`:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
version: '3.8'
|
||||||
|
|
||||||
|
services:
|
||||||
|
gitea-mcp-wrapper:
|
||||||
|
build:
|
||||||
|
context: .
|
||||||
|
dockerfile: Dockerfile
|
||||||
|
image: gitea-mcp-wrapper:latest
|
||||||
|
container_name: gitea-mcp-wrapper
|
||||||
|
restart: always
|
||||||
|
ports:
|
||||||
|
- "127.0.0.1:8000:8000" # Bind to localhost only
|
||||||
|
environment:
|
||||||
|
- GITEA_URL=${GITEA_URL}
|
||||||
|
- GITEA_TOKEN=${GITEA_TOKEN}
|
||||||
|
- GITEA_OWNER=${GITEA_OWNER}
|
||||||
|
- GITEA_REPO=${GITEA_REPO}
|
||||||
|
- HTTP_HOST=0.0.0.0
|
||||||
|
- HTTP_PORT=8000
|
||||||
|
- AUTH_TOKEN=${AUTH_TOKEN}
|
||||||
|
healthcheck:
|
||||||
|
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()"]
|
||||||
|
interval: 30s
|
||||||
|
timeout: 3s
|
||||||
|
retries: 3
|
||||||
|
start_period: 5s
|
||||||
|
logging:
|
||||||
|
driver: "json-file"
|
||||||
|
options:
|
||||||
|
max-size: "10m"
|
||||||
|
max-file: "3"
|
||||||
|
networks:
|
||||||
|
- gitea-mcp-network
|
||||||
|
|
||||||
|
networks:
|
||||||
|
gitea-mcp-network:
|
||||||
|
driver: bridge
|
||||||
|
```
|
||||||
|
|
||||||
|
### Docker Build Options
|
||||||
|
|
||||||
|
**Build the image:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker build -t gitea-mcp-wrapper:latest .
|
||||||
|
```
|
||||||
|
|
||||||
|
**Build with specific Python version:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker build --build-arg PYTHON_VERSION=3.11 -t gitea-mcp-wrapper:latest .
|
||||||
|
```
|
||||||
|
|
||||||
|
**Tag for registry:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker tag gitea-mcp-wrapper:latest registry.example.com/gitea-mcp-wrapper:latest
|
||||||
|
docker push registry.example.com/gitea-mcp-wrapper:latest
|
||||||
|
```
|
||||||
|
|
||||||
|
## Security Best Practices
|
||||||
|
|
||||||
|
### 1. Use Authentication
|
||||||
|
|
||||||
|
Always set `AUTH_TOKEN` in production:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Generate a secure token
|
||||||
|
openssl rand -base64 32
|
||||||
|
|
||||||
|
# Add to .env
|
||||||
|
AUTH_TOKEN=<generated_token>
|
||||||
|
```
|
||||||
|
|
||||||
|
### 2. Use HTTPS
|
||||||
|
|
||||||
|
Never expose the wrapper directly to the internet without HTTPS. Use a reverse proxy (see below).
|
||||||
|
|
||||||
|
### 3. Network Isolation
|
||||||
|
|
||||||
|
- Bind to localhost only (`127.0.0.1`) if using a reverse proxy
|
||||||
|
- Use Docker networks to isolate services
|
||||||
|
- Consider VPN or private networking for access
|
||||||
|
|
||||||
|
### 4. Secrets Management
|
||||||
|
|
||||||
|
Don't use `.env` files in production. Use Docker secrets, Kubernetes secrets, or a secrets manager:
|
||||||
|
|
||||||
|
**Docker Secrets Example:**
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
services:
|
||||||
|
gitea-mcp-wrapper:
|
||||||
|
secrets:
|
||||||
|
- gitea_token
|
||||||
|
- auth_token
|
||||||
|
environment:
|
||||||
|
- GITEA_TOKEN_FILE=/run/secrets/gitea_token
|
||||||
|
- AUTH_TOKEN_FILE=/run/secrets/auth_token
|
||||||
|
|
||||||
|
secrets:
|
||||||
|
gitea_token:
|
||||||
|
external: true
|
||||||
|
auth_token:
|
||||||
|
external: true
|
||||||
|
```
|
||||||
|
|
||||||
|
### 5. Regular Updates
|
||||||
|
|
||||||
|
- Rotate Gitea API token regularly
|
||||||
|
- Rotate AUTH_TOKEN regularly
|
||||||
|
- Keep Docker base image updated
|
||||||
|
- Update dependencies: `pip install --upgrade -r requirements.txt`
|
||||||
|
|
||||||
|
### 6. Minimal Permissions
|
||||||
|
|
||||||
|
Grant the Gitea API token only the minimum required permissions:
|
||||||
|
|
||||||
|
- Repository read/write
|
||||||
|
- Issue management
|
||||||
|
- Label management
|
||||||
|
- Milestone management
|
||||||
|
|
||||||
|
Avoid granting admin or organization-level permissions.
|
||||||
|
|
||||||
|
## Monitoring and Health Checks
|
||||||
|
|
||||||
|
### Health Check Endpoints
|
||||||
|
|
||||||
|
The wrapper provides three health check endpoints:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
GET /health
|
||||||
|
GET /healthz
|
||||||
|
GET /ping
|
||||||
|
```
|
||||||
|
|
||||||
|
All return `{"status": "healthy"}` with HTTP 200 when the server is operational.
|
||||||
|
|
||||||
|
### Docker Health Checks
|
||||||
|
|
||||||
|
Docker automatically monitors the health check and can restart if unhealthy:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
healthcheck:
|
||||||
|
test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
|
||||||
|
interval: 30s
|
||||||
|
timeout: 3s
|
||||||
|
retries: 3
|
||||||
|
start_period: 5s
|
||||||
|
```
|
||||||
|
|
||||||
|
### Monitoring Integration
|
||||||
|
|
||||||
|
**Prometheus metrics:** (Not yet implemented, but can be added)
|
||||||
|
|
||||||
|
**Log monitoring:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# View logs
|
||||||
|
docker-compose logs -f gitea-mcp-wrapper
|
||||||
|
|
||||||
|
# JSON structured logs
|
||||||
|
docker logs gitea-mcp-wrapper --tail 100
|
||||||
|
```
|
||||||
|
|
||||||
|
**Uptime monitoring:**
|
||||||
|
|
||||||
|
Use tools like UptimeRobot, Pingdom, or Datadog to monitor `/health` endpoint.
|
||||||
|
|
||||||
|
## Reverse Proxy Configuration
|
||||||
|
|
||||||
|
### Nginx
|
||||||
|
|
||||||
|
```nginx
|
||||||
|
server {
|
||||||
|
listen 443 ssl http2;
|
||||||
|
server_name mcp.example.com;
|
||||||
|
|
||||||
|
ssl_certificate /etc/nginx/ssl/cert.pem;
|
||||||
|
ssl_certificate_key /etc/nginx/ssl/key.pem;
|
||||||
|
|
||||||
|
location / {
|
||||||
|
proxy_pass http://127.0.0.1:8000;
|
||||||
|
proxy_set_header Host $host;
|
||||||
|
proxy_set_header X-Real-IP $remote_addr;
|
||||||
|
proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
|
||||||
|
proxy_set_header X-Forwarded-Proto $scheme;
|
||||||
|
|
||||||
|
# Pass through Authorization header
|
||||||
|
proxy_set_header Authorization $http_authorization;
|
||||||
|
proxy_pass_header Authorization;
|
||||||
|
|
||||||
|
# WebSocket support (if needed in future)
|
||||||
|
proxy_http_version 1.1;
|
||||||
|
proxy_set_header Upgrade $http_upgrade;
|
||||||
|
proxy_set_header Connection "upgrade";
|
||||||
|
|
||||||
|
# Timeouts
|
||||||
|
proxy_connect_timeout 60s;
|
||||||
|
proxy_send_timeout 60s;
|
||||||
|
proxy_read_timeout 60s;
|
||||||
|
}
|
||||||
|
|
||||||
|
# Health check endpoint (optional, can bypass auth)
|
||||||
|
location /health {
|
||||||
|
proxy_pass http://127.0.0.1:8000/health;
|
||||||
|
access_log off;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Caddy
|
||||||
|
|
||||||
|
```caddyfile
|
||||||
|
mcp.example.com {
|
||||||
|
reverse_proxy localhost:8000 {
|
||||||
|
# Pass through Authorization header
|
||||||
|
header_up Authorization {>Authorization}
|
||||||
|
}
|
||||||
|
|
||||||
|
# Optional: Rate limiting
|
||||||
|
rate_limit {
|
||||||
|
zone mcp_zone
|
||||||
|
rate 100r/m
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### Traefik
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
# docker-compose.yml
|
||||||
|
services:
|
||||||
|
gitea-mcp-wrapper:
|
||||||
|
labels:
|
||||||
|
- "traefik.enable=true"
|
||||||
|
- "traefik.http.routers.mcp.rule=Host(`mcp.example.com`)"
|
||||||
|
- "traefik.http.routers.mcp.entrypoints=websecure"
|
||||||
|
- "traefik.http.routers.mcp.tls.certresolver=letsencrypt"
|
||||||
|
- "traefik.http.services.mcp.loadbalancer.server.port=8000"
|
||||||
|
```
|
||||||
|
|
||||||
|
## Cloud Deployment
|
||||||
|
|
||||||
|
### AWS EC2
|
||||||
|
|
||||||
|
1. **Launch EC2 instance:**
|
||||||
|
- Amazon Linux 2 or Ubuntu 22.04
|
||||||
|
- t3.micro or larger
|
||||||
|
- Security group: Allow port 443 (HTTPS)
|
||||||
|
|
||||||
|
2. **Install Docker:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
sudo yum update -y
|
||||||
|
sudo yum install -y docker
|
||||||
|
sudo service docker start
|
||||||
|
sudo usermod -aG docker ec2-user
|
||||||
|
```
|
||||||
|
|
||||||
|
3. **Deploy wrapper:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
git clone https://github.com/lmiranda/gitea-mcp-remote.git
|
||||||
|
cd gitea-mcp-remote
|
||||||
|
cp .env.docker.example .env
|
||||||
|
nano .env # Configure
|
||||||
|
docker-compose up -d
|
||||||
|
```
|
||||||
|
|
||||||
|
4. **Configure Nginx or ALB for HTTPS**
|
||||||
|
|
||||||
|
### AWS ECS (Fargate)
|
||||||
|
|
||||||
|
1. **Create task definition:**
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"family": "gitea-mcp-wrapper",
|
||||||
|
"networkMode": "awsvpc",
|
||||||
|
"requiresCompatibilities": ["FARGATE"],
|
||||||
|
"cpu": "256",
|
||||||
|
"memory": "512",
|
||||||
|
"containerDefinitions": [
|
||||||
|
{
|
||||||
|
"name": "gitea-mcp-wrapper",
|
||||||
|
"image": "your-ecr-repo/gitea-mcp-wrapper:latest",
|
||||||
|
"portMappings": [
|
||||||
|
{
|
||||||
|
"containerPort": 8000,
|
||||||
|
"protocol": "tcp"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"environment": [
|
||||||
|
{"name": "GITEA_URL", "value": "https://gitea.example.com"},
|
||||||
|
{"name": "HTTP_HOST", "value": "0.0.0.0"},
|
||||||
|
{"name": "HTTP_PORT", "value": "8000"}
|
||||||
|
],
|
||||||
|
"secrets": [
|
||||||
|
{
|
||||||
|
"name": "GITEA_TOKEN",
|
||||||
|
"valueFrom": "arn:aws:secretsmanager:region:account:secret:gitea-token"
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"name": "AUTH_TOKEN",
|
||||||
|
"valueFrom": "arn:aws:secretsmanager:region:account:secret:auth-token"
|
||||||
|
}
|
||||||
|
],
|
||||||
|
"healthCheck": {
|
||||||
|
"command": ["CMD-SHELL", "curl -f http://localhost:8000/health || exit 1"],
|
||||||
|
"interval": 30,
|
||||||
|
"timeout": 5,
|
||||||
|
"retries": 3,
|
||||||
|
"startPeriod": 10
|
||||||
|
}
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
2. **Create ECS service with ALB**
|
||||||
|
|
||||||
|
### Google Cloud Run
|
||||||
|
|
||||||
|
1. **Build and push image:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
gcloud builds submit --tag gcr.io/PROJECT_ID/gitea-mcp-wrapper
|
||||||
|
```
|
||||||
|
|
||||||
|
2. **Deploy:**
|
||||||
|
|
||||||
|
```bash
|
||||||
|
gcloud run deploy gitea-mcp-wrapper \
|
||||||
|
--image gcr.io/PROJECT_ID/gitea-mcp-wrapper \
|
||||||
|
--platform managed \
|
||||||
|
--region us-central1 \
|
||||||
|
--allow-unauthenticated \
|
||||||
|
--set-env-vars GITEA_URL=https://gitea.example.com \
|
||||||
|
--set-secrets GITEA_TOKEN=gitea-token:latest,AUTH_TOKEN=auth-token:latest \
|
||||||
|
--port 8000
|
||||||
|
```
|
||||||
|
|
||||||
|
### Azure Container Instances
|
||||||
|
|
||||||
|
```bash
|
||||||
|
az container create \
|
||||||
|
--resource-group myResourceGroup \
|
||||||
|
--name gitea-mcp-wrapper \
|
||||||
|
--image your-registry/gitea-mcp-wrapper:latest \
|
||||||
|
--ports 8000 \
|
||||||
|
--dns-name-label gitea-mcp \
|
||||||
|
--environment-variables \
|
||||||
|
GITEA_URL=https://gitea.example.com \
|
||||||
|
HTTP_HOST=0.0.0.0 \
|
||||||
|
HTTP_PORT=8000 \
|
||||||
|
--secure-environment-variables \
|
||||||
|
GITEA_TOKEN=your_token \
|
||||||
|
AUTH_TOKEN=your_auth_token
|
||||||
|
```
|
||||||
|
|
||||||
|
## Kubernetes Deployment
|
||||||
|
|
||||||
|
### Deployment Manifest
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
apiVersion: apps/v1
|
||||||
|
kind: Deployment
|
||||||
|
metadata:
|
||||||
|
name: gitea-mcp-wrapper
|
||||||
|
namespace: default
|
||||||
|
spec:
|
||||||
|
replicas: 2
|
||||||
|
selector:
|
||||||
|
matchLabels:
|
||||||
|
app: gitea-mcp-wrapper
|
||||||
|
template:
|
||||||
|
metadata:
|
||||||
|
labels:
|
||||||
|
app: gitea-mcp-wrapper
|
||||||
|
spec:
|
||||||
|
containers:
|
||||||
|
- name: gitea-mcp-wrapper
|
||||||
|
image: your-registry/gitea-mcp-wrapper:latest
|
||||||
|
ports:
|
||||||
|
- containerPort: 8000
|
||||||
|
env:
|
||||||
|
- name: GITEA_URL
|
||||||
|
value: "https://gitea.example.com"
|
||||||
|
- name: HTTP_HOST
|
||||||
|
value: "0.0.0.0"
|
||||||
|
- name: HTTP_PORT
|
||||||
|
value: "8000"
|
||||||
|
- name: GITEA_TOKEN
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: gitea-secrets
|
||||||
|
key: token
|
||||||
|
- name: AUTH_TOKEN
|
||||||
|
valueFrom:
|
||||||
|
secretKeyRef:
|
||||||
|
name: gitea-secrets
|
||||||
|
key: auth-token
|
||||||
|
livenessProbe:
|
||||||
|
httpGet:
|
||||||
|
path: /health
|
||||||
|
port: 8000
|
||||||
|
initialDelaySeconds: 10
|
||||||
|
periodSeconds: 30
|
||||||
|
readinessProbe:
|
||||||
|
httpGet:
|
||||||
|
path: /health
|
||||||
|
port: 8000
|
||||||
|
initialDelaySeconds: 5
|
||||||
|
periodSeconds: 10
|
||||||
|
resources:
|
||||||
|
requests:
|
||||||
|
memory: "128Mi"
|
||||||
|
cpu: "100m"
|
||||||
|
limits:
|
||||||
|
memory: "256Mi"
|
||||||
|
cpu: "200m"
|
||||||
|
---
|
||||||
|
apiVersion: v1
|
||||||
|
kind: Service
|
||||||
|
metadata:
|
||||||
|
name: gitea-mcp-wrapper
|
||||||
|
namespace: default
|
||||||
|
spec:
|
||||||
|
selector:
|
||||||
|
app: gitea-mcp-wrapper
|
||||||
|
ports:
|
||||||
|
- protocol: TCP
|
||||||
|
port: 80
|
||||||
|
targetPort: 8000
|
||||||
|
type: ClusterIP
|
||||||
|
---
|
||||||
|
apiVersion: networking.k8s.io/v1
|
||||||
|
kind: Ingress
|
||||||
|
metadata:
|
||||||
|
name: gitea-mcp-wrapper
|
||||||
|
namespace: default
|
||||||
|
annotations:
|
||||||
|
cert-manager.io/cluster-issuer: "letsencrypt-prod"
|
||||||
|
spec:
|
||||||
|
tls:
|
||||||
|
- hosts:
|
||||||
|
- mcp.example.com
|
||||||
|
secretName: mcp-tls
|
||||||
|
rules:
|
||||||
|
- host: mcp.example.com
|
||||||
|
http:
|
||||||
|
paths:
|
||||||
|
- path: /
|
||||||
|
pathType: Prefix
|
||||||
|
backend:
|
||||||
|
service:
|
||||||
|
name: gitea-mcp-wrapper
|
||||||
|
port:
|
||||||
|
number: 80
|
||||||
|
```
|
||||||
|
|
||||||
|
### Secrets Management
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Create secret
|
||||||
|
kubectl create secret generic gitea-secrets \
|
||||||
|
--from-literal=token=your_gitea_token \
|
||||||
|
--from-literal=auth-token=your_auth_token \
|
||||||
|
--namespace=default
|
||||||
|
```
|
||||||
|
|
||||||
|
## Troubleshooting
|
||||||
|
|
||||||
|
### Container Won't Start
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Check logs
|
||||||
|
docker-compose logs gitea-mcp-wrapper
|
||||||
|
|
||||||
|
# Check container status
|
||||||
|
docker-compose ps
|
||||||
|
|
||||||
|
# Rebuild image
|
||||||
|
docker-compose build --no-cache
|
||||||
|
docker-compose up -d
|
||||||
|
```
|
||||||
|
|
||||||
|
### Health Check Failing
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Test health endpoint directly
|
||||||
|
docker exec gitea-mcp-wrapper curl http://localhost:8000/health
|
||||||
|
|
||||||
|
# Check if server is listening
|
||||||
|
docker exec gitea-mcp-wrapper netstat -tlnp
|
||||||
|
```
|
||||||
|
|
||||||
|
### Cannot Reach Gitea from Container
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Test connectivity
|
||||||
|
docker exec gitea-mcp-wrapper curl -v https://gitea.example.com
|
||||||
|
|
||||||
|
# Check DNS resolution
|
||||||
|
docker exec gitea-mcp-wrapper nslookup gitea.example.com
|
||||||
|
|
||||||
|
# For docker-compose, ensure network allows egress
|
||||||
|
```
|
||||||
|
|
||||||
|
### High Memory Usage
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Check container stats
|
||||||
|
docker stats gitea-mcp-wrapper
|
||||||
|
|
||||||
|
# Adjust resource limits in docker-compose.yml
|
||||||
|
deploy:
|
||||||
|
resources:
|
||||||
|
limits:
|
||||||
|
memory: 256M
|
||||||
|
```
|
||||||
|
|
||||||
|
### Authentication Failures
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Verify AUTH_TOKEN is set
|
||||||
|
docker exec gitea-mcp-wrapper printenv AUTH_TOKEN
|
||||||
|
|
||||||
|
# Test with curl
|
||||||
|
curl -H "Authorization: Bearer YOUR_TOKEN" http://localhost:8000/tools/list
|
||||||
|
|
||||||
|
# Check logs for auth failures
|
||||||
|
docker-compose logs gitea-mcp-wrapper | grep -i auth
|
||||||
|
```
|
||||||
|
|
||||||
|
## Scaling Considerations
|
||||||
|
|
||||||
|
### Horizontal Scaling
|
||||||
|
|
||||||
|
The wrapper is stateless and can be scaled horizontally:
|
||||||
|
|
||||||
|
```yaml
|
||||||
|
# docker-compose.yml
|
||||||
|
services:
|
||||||
|
gitea-mcp-wrapper:
|
||||||
|
deploy:
|
||||||
|
replicas: 3
|
||||||
|
```
|
||||||
|
|
||||||
|
Or in Kubernetes:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
kubectl scale deployment gitea-mcp-wrapper --replicas=5
|
||||||
|
```
|
||||||
|
|
||||||
|
### Load Balancing
|
||||||
|
|
||||||
|
Use a load balancer to distribute traffic:
|
||||||
|
- Docker Swarm: Built-in load balancing
|
||||||
|
- Kubernetes: Service with multiple pods
|
||||||
|
- Cloud: AWS ALB, GCP Load Balancer, Azure Load Balancer
|
||||||
|
|
||||||
|
### Caching
|
||||||
|
|
||||||
|
Consider caching responses to reduce Gitea API load:
|
||||||
|
- Add Redis or Memcached
|
||||||
|
- Cache tool list responses
|
||||||
|
- Cache frequently accessed issues/labels
|
||||||
|
|
||||||
|
### Rate Limiting
|
||||||
|
|
||||||
|
Implement rate limiting at reverse proxy level to prevent API abuse:
|
||||||
|
|
||||||
|
**Nginx:**
|
||||||
|
```nginx
|
||||||
|
limit_req_zone $binary_remote_addr zone=mcp:10m rate=10r/s;
|
||||||
|
limit_req zone=mcp burst=20 nodelay;
|
||||||
|
```
|
||||||
|
|
||||||
|
**Caddy:**
|
||||||
|
```caddyfile
|
||||||
|
rate_limit {
|
||||||
|
rate 100r/m
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Backup and Disaster Recovery
|
||||||
|
|
||||||
|
### Configuration Backup
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Backup .env file
|
||||||
|
cp .env .env.backup.$(date +%Y%m%d)
|
||||||
|
|
||||||
|
# Backup docker-compose.yml
|
||||||
|
cp docker-compose.yml docker-compose.yml.backup.$(date +%Y%m%d)
|
||||||
|
```
|
||||||
|
|
||||||
|
### Image Backup
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Save Docker image
|
||||||
|
docker save gitea-mcp-wrapper:latest | gzip > gitea-mcp-wrapper-backup.tar.gz
|
||||||
|
|
||||||
|
# Load Docker image
|
||||||
|
docker load < gitea-mcp-wrapper-backup.tar.gz
|
||||||
|
```
|
||||||
|
|
||||||
|
### Recovery Plan
|
||||||
|
|
||||||
|
1. Restore configuration files
|
||||||
|
2. Rebuild or load Docker image
|
||||||
|
3. Start services: `docker-compose up -d`
|
||||||
|
4. Verify health: `curl http://localhost:8000/health`
|
||||||
|
5. Test authentication and tool access
|
||||||
|
|
||||||
|
## Production Checklist
|
||||||
|
|
||||||
|
- [ ] HTTPS configured via reverse proxy
|
||||||
|
- [ ] `AUTH_TOKEN` set and secure
|
||||||
|
- [ ] Secrets stored in secrets manager (not `.env`)
|
||||||
|
- [ ] Health checks configured
|
||||||
|
- [ ] Monitoring and alerting set up
|
||||||
|
- [ ] Logs aggregated and retained
|
||||||
|
- [ ] Firewall rules configured
|
||||||
|
- [ ] Rate limiting enabled
|
||||||
|
- [ ] Resource limits set
|
||||||
|
- [ ] Backup strategy in place
|
||||||
|
- [ ] Disaster recovery plan documented
|
||||||
|
- [ ] Security updates scheduled
|
||||||
|
- [ ] Token rotation process defined
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
|
**For questions or issues, please open an issue on the [GitHub repository](https://github.com/lmiranda/gitea-mcp-remote/issues).**
|
||||||
54
Dockerfile
Normal file
54
Dockerfile
Normal file
@@ -0,0 +1,54 @@
|
|||||||
|
# Gitea HTTP MCP Wrapper Dockerfile
|
||||||
|
# Multi-stage build for optimized image size
|
||||||
|
|
||||||
|
FROM python:3.11-slim as builder
|
||||||
|
|
||||||
|
# Set working directory
|
||||||
|
WORKDIR /build
|
||||||
|
|
||||||
|
# Install build dependencies
|
||||||
|
RUN apt-get update && \
|
||||||
|
apt-get install -y --no-install-recommends \
|
||||||
|
gcc \
|
||||||
|
&& rm -rf /var/lib/apt/lists/*
|
||||||
|
|
||||||
|
# Copy requirements first for better caching
|
||||||
|
COPY requirements.txt .
|
||||||
|
RUN pip install --user --no-cache-dir -r requirements.txt
|
||||||
|
|
||||||
|
# Copy source code
|
||||||
|
COPY pyproject.toml .
|
||||||
|
COPY src/ src/
|
||||||
|
|
||||||
|
# Install package
|
||||||
|
RUN pip install --user --no-cache-dir -e .
|
||||||
|
|
||||||
|
# Production stage
|
||||||
|
FROM python:3.11-slim
|
||||||
|
|
||||||
|
# Set working directory
|
||||||
|
WORKDIR /app
|
||||||
|
|
||||||
|
# Copy installed packages from builder
|
||||||
|
COPY --from=builder /root/.local /root/.local
|
||||||
|
|
||||||
|
# Copy source code
|
||||||
|
COPY src/ src/
|
||||||
|
COPY pyproject.toml .
|
||||||
|
|
||||||
|
# Make sure scripts in .local are usable
|
||||||
|
ENV PATH=/root/.local/bin:$PATH
|
||||||
|
|
||||||
|
# Set Python environment variables
|
||||||
|
ENV PYTHONUNBUFFERED=1
|
||||||
|
ENV PYTHONDONTWRITEBYTECODE=1
|
||||||
|
|
||||||
|
# Expose default port
|
||||||
|
EXPOSE 8000
|
||||||
|
|
||||||
|
# Health check
|
||||||
|
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
|
||||||
|
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()"
|
||||||
|
|
||||||
|
# Run the HTTP MCP server
|
||||||
|
CMD ["gitea-http-wrapper"]
|
||||||
345
README.md
345
README.md
@@ -1,27 +1,83 @@
|
|||||||
# Gitea MCP Remote
|
# Gitea HTTP MCP Wrapper
|
||||||
|
|
||||||
MCP server for Gitea API integration.
|
An HTTP transport wrapper around the official Gitea MCP server that enables AI assistants like Claude Desktop to interact with Gitea repositories via HTTP. This wrapper provides authentication, tool filtering, and HTTP transport while delegating Gitea operations to the official `gitea-mcp-server`.
|
||||||
|
|
||||||
## Overview
|
## Architecture
|
||||||
|
|
||||||
This project provides a Model Context Protocol (MCP) server that enables AI assistants to interact with Gitea through its API.
|
This is NOT a standalone MCP server. It's an HTTP wrapper that:
|
||||||
|
1. Wraps the official `gitea-mcp-server` (stdio transport)
|
||||||
|
2. Provides HTTP transport for Claude Desktop compatibility
|
||||||
|
3. Adds Bearer token authentication
|
||||||
|
4. Filters tools for Claude Desktop compatibility
|
||||||
|
5. Proxies requests between HTTP and stdio transport
|
||||||
|
|
||||||
## Project Status
|
```
|
||||||
|
Claude Desktop (HTTP) → HTTP Wrapper → Gitea MCP Server (stdio) → Gitea API
|
||||||
|
```
|
||||||
|
|
||||||
Currently in initial development. Project structure has been initialized.
|
## Features
|
||||||
|
|
||||||
|
- **HTTP Transport**: Exposes MCP server via HTTP for Claude Desktop
|
||||||
|
- **Authentication**: Optional Bearer token authentication
|
||||||
|
- **Tool Filtering**: Enable/disable specific tools for compatibility
|
||||||
|
- **Docker Deployment**: Production-ready containerization
|
||||||
|
- **Health Checks**: Monitoring endpoints (`/health`, `/healthz`, `/ping`)
|
||||||
|
- **Async Architecture**: Built on Starlette and uvicorn
|
||||||
|
|
||||||
## Requirements
|
## Requirements
|
||||||
|
|
||||||
- Python >= 3.10
|
- Python >= 3.10
|
||||||
|
- Official `gitea-mcp-server` package (auto-installed as dependency)
|
||||||
- Gitea instance with API access
|
- Gitea instance with API access
|
||||||
|
- Gitea API token with appropriate permissions
|
||||||
|
|
||||||
|
## Quick Start with Docker
|
||||||
|
|
||||||
|
The easiest way to deploy is using Docker Compose:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# 1. Clone the repository
|
||||||
|
git clone https://github.com/lmiranda/gitea-mcp-remote.git
|
||||||
|
cd gitea-mcp-remote
|
||||||
|
|
||||||
|
# 2. Create .env file from template
|
||||||
|
cp .env.docker.example .env
|
||||||
|
|
||||||
|
# 3. Edit .env with your Gitea credentials
|
||||||
|
nano .env
|
||||||
|
|
||||||
|
# 4. Start the server
|
||||||
|
docker-compose up -d
|
||||||
|
|
||||||
|
# 5. Check health
|
||||||
|
curl http://localhost:8000/health
|
||||||
|
```
|
||||||
|
|
||||||
|
The server will be available at `http://localhost:8000`.
|
||||||
|
|
||||||
|
See [DEPLOYMENT.md](DEPLOYMENT.md) for detailed deployment instructions.
|
||||||
|
|
||||||
## Installation
|
## Installation
|
||||||
|
|
||||||
|
### Option 1: Docker (Recommended)
|
||||||
|
|
||||||
|
See [Quick Start](#quick-start-with-docker) above or [DEPLOYMENT.md](DEPLOYMENT.md).
|
||||||
|
|
||||||
|
### Option 2: From Source
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
|
# Clone the repository
|
||||||
|
git clone https://github.com/lmiranda/gitea-mcp-remote.git
|
||||||
|
cd gitea-mcp-remote
|
||||||
|
|
||||||
|
# Install the wrapper and its dependencies (including gitea-mcp-server)
|
||||||
pip install -e .
|
pip install -e .
|
||||||
|
|
||||||
|
# Or use requirements.txt
|
||||||
|
pip install -r requirements.txt
|
||||||
```
|
```
|
||||||
|
|
||||||
## Development
|
### For Development
|
||||||
|
|
||||||
Install with development dependencies:
|
Install with development dependencies:
|
||||||
|
|
||||||
@@ -29,12 +85,281 @@ Install with development dependencies:
|
|||||||
pip install -e ".[dev]"
|
pip install -e ".[dev]"
|
||||||
```
|
```
|
||||||
|
|
||||||
Run tests:
|
## Configuration
|
||||||
|
|
||||||
|
The wrapper uses environment variables or a `.env` file for configuration.
|
||||||
|
|
||||||
|
### Required Configuration
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
pytest
|
# Gitea Instance
|
||||||
|
GITEA_URL=https://gitea.example.com
|
||||||
|
GITEA_TOKEN=your_gitea_api_token_here
|
||||||
|
GITEA_OWNER=your_username_or_org
|
||||||
|
GITEA_REPO=your_repo_name
|
||||||
|
|
||||||
|
# HTTP Server
|
||||||
|
HTTP_HOST=127.0.0.1 # Use 0.0.0.0 in Docker
|
||||||
|
HTTP_PORT=8000
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### Optional Configuration
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Bearer Authentication (optional but recommended)
|
||||||
|
AUTH_TOKEN=your_secret_bearer_token
|
||||||
|
|
||||||
|
# Tool Filtering (optional)
|
||||||
|
ENABLED_TOOLS=list_issues,create_issue,update_issue # Whitelist mode
|
||||||
|
# OR
|
||||||
|
DISABLED_TOOLS=delete_issue,close_milestone # Blacklist mode
|
||||||
|
```
|
||||||
|
|
||||||
|
### Getting a Gitea API Token
|
||||||
|
|
||||||
|
1. Log into your Gitea instance
|
||||||
|
2. Navigate to Settings > Applications
|
||||||
|
3. Under "Generate New Token", enter a name (e.g., "MCP Wrapper")
|
||||||
|
4. Select appropriate permissions (minimum: read/write for repositories)
|
||||||
|
5. Click "Generate Token" and copy the token
|
||||||
|
6. Add the token to your `.env` file
|
||||||
|
|
||||||
|
## Usage
|
||||||
|
|
||||||
|
### Running the Server
|
||||||
|
|
||||||
|
#### With Docker
|
||||||
|
|
||||||
|
```bash
|
||||||
|
docker-compose up -d
|
||||||
|
```
|
||||||
|
|
||||||
|
#### From Source
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Create .env file from template
|
||||||
|
cp .env.example .env
|
||||||
|
# Edit .env with your configuration
|
||||||
|
nano .env
|
||||||
|
|
||||||
|
# Run the server
|
||||||
|
gitea-http-wrapper
|
||||||
|
```
|
||||||
|
|
||||||
|
The server will start on the configured host/port (default: `http://127.0.0.1:8000`).
|
||||||
|
|
||||||
|
### HTTP Endpoints
|
||||||
|
|
||||||
|
#### Health Check
|
||||||
|
```bash
|
||||||
|
GET /health
|
||||||
|
GET /healthz
|
||||||
|
GET /ping
|
||||||
|
|
||||||
|
Response: {"status": "healthy"}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### List Tools
|
||||||
|
```bash
|
||||||
|
POST /tools/list
|
||||||
|
|
||||||
|
Response: {
|
||||||
|
"tools": [
|
||||||
|
{"name": "list_issues", "description": "...", "inputSchema": {...}},
|
||||||
|
...
|
||||||
|
]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### Call Tool
|
||||||
|
```bash
|
||||||
|
POST /tools/call
|
||||||
|
Content-Type: application/json
|
||||||
|
Authorization: Bearer YOUR_TOKEN # If auth enabled
|
||||||
|
|
||||||
|
{
|
||||||
|
"name": "list_issues",
|
||||||
|
"arguments": {
|
||||||
|
"owner": "myorg",
|
||||||
|
"repo": "myrepo",
|
||||||
|
"state": "open"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### With Claude Desktop
|
||||||
|
|
||||||
|
Configure Claude Desktop to use the HTTP wrapper:
|
||||||
|
|
||||||
|
**Location:**
|
||||||
|
- macOS: `~/Library/Application Support/Claude/claude_desktop_config.json`
|
||||||
|
- Windows: `%APPDATA%\Claude\claude_desktop_config.json`
|
||||||
|
- Linux: `~/.config/Claude/claude_desktop_config.json`
|
||||||
|
|
||||||
|
**Configuration:**
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"mcpServers": {
|
||||||
|
"gitea": {
|
||||||
|
"url": "http://localhost:8000",
|
||||||
|
"headers": {
|
||||||
|
"Authorization": "Bearer YOUR_TOKEN"
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Available Tools
|
||||||
|
|
||||||
|
The wrapper exposes all tools from the official `gitea-mcp-server`. See the [official Gitea MCP documentation](https://github.com/modelcontextprotocol/servers/tree/main/src/gitea) for the complete list of available tools:
|
||||||
|
|
||||||
|
- **Issues**: List, get, create, update issues
|
||||||
|
- **Labels**: List, create labels
|
||||||
|
- **Milestones**: List, create milestones
|
||||||
|
|
||||||
|
Tool availability can be controlled via the `ENABLED_TOOLS` or `DISABLED_TOOLS` configuration.
|
||||||
|
|
||||||
|
## Development
|
||||||
|
|
||||||
|
### Setup Development Environment
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Install with development dependencies
|
||||||
|
pip install -e ".[dev]"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Running Tests
|
||||||
|
|
||||||
|
```bash
|
||||||
|
# Run all tests
|
||||||
|
pytest
|
||||||
|
|
||||||
|
# Run with coverage
|
||||||
|
pytest --cov=gitea_http_wrapper
|
||||||
|
|
||||||
|
# Run specific test file
|
||||||
|
pytest src/gitea_http_wrapper/tests/test_config.py
|
||||||
|
```
|
||||||
|
|
||||||
|
### Project Structure
|
||||||
|
|
||||||
|
```
|
||||||
|
gitea-mcp-remote/
|
||||||
|
├── src/
|
||||||
|
│ └── gitea_http_wrapper/
|
||||||
|
│ ├── __init__.py
|
||||||
|
│ ├── server.py # Main HTTP server
|
||||||
|
│ ├── config/
|
||||||
|
│ │ ├── __init__.py
|
||||||
|
│ │ └── settings.py # Configuration loader
|
||||||
|
│ ├── middleware/
|
||||||
|
│ │ ├── __init__.py
|
||||||
|
│ │ └── auth.py # HTTP authentication
|
||||||
|
│ ├── filtering/
|
||||||
|
│ │ ├── __init__.py
|
||||||
|
│ │ └── filter.py # Tool filtering
|
||||||
|
│ └── tests/ # Test suite
|
||||||
|
│ ├── conftest.py
|
||||||
|
│ ├── test_config.py
|
||||||
|
│ ├── test_filtering.py
|
||||||
|
│ └── test_middleware.py
|
||||||
|
├── Dockerfile # Docker image
|
||||||
|
├── docker-compose.yml # Docker orchestration
|
||||||
|
├── pyproject.toml # Project config
|
||||||
|
├── requirements.txt # Dependencies
|
||||||
|
├── .env.example # Config template
|
||||||
|
├── .env.docker.example # Docker config template
|
||||||
|
├── README.md # This file
|
||||||
|
└── DEPLOYMENT.md # Deployment guide
|
||||||
|
```
|
||||||
|
|
||||||
|
## Deployment
|
||||||
|
|
||||||
|
For production deployment instructions, see [DEPLOYMENT.md](DEPLOYMENT.md), which covers:
|
||||||
|
|
||||||
|
- Docker deployment
|
||||||
|
- Docker Compose orchestration
|
||||||
|
- Security best practices
|
||||||
|
- Monitoring and health checks
|
||||||
|
- Scaling considerations
|
||||||
|
- Cloud deployment (AWS, GCP, Azure)
|
||||||
|
- Kubernetes deployment
|
||||||
|
|
||||||
|
## Troubleshooting
|
||||||
|
|
||||||
|
### Authentication Errors
|
||||||
|
|
||||||
|
If you receive authentication errors:
|
||||||
|
|
||||||
|
1. Verify your `GITEA_TOKEN` is correct
|
||||||
|
2. Check that the token has appropriate permissions
|
||||||
|
3. Ensure your `GITEA_URL` does NOT include `/api/v1` (wrapper adds it)
|
||||||
|
4. Verify the Gitea instance is accessible from the wrapper's network
|
||||||
|
|
||||||
|
### HTTP 401/403 Errors
|
||||||
|
|
||||||
|
If Claude Desktop receives 401 or 403 errors:
|
||||||
|
|
||||||
|
1. Check that `AUTH_TOKEN` is configured (if authentication is enabled)
|
||||||
|
2. Verify Claude Desktop config includes the correct `Authorization` header
|
||||||
|
3. Check server logs for authentication failures
|
||||||
|
|
||||||
|
### Connection Errors
|
||||||
|
|
||||||
|
If the wrapper cannot connect to Gitea:
|
||||||
|
|
||||||
|
1. Check that `GITEA_URL` is correct and accessible
|
||||||
|
2. Verify network connectivity to the Gitea instance
|
||||||
|
3. Check for firewalls or proxies blocking the connection
|
||||||
|
4. In Docker: Ensure the container can reach the Gitea host
|
||||||
|
|
||||||
|
### gitea-mcp-server Not Found
|
||||||
|
|
||||||
|
If the wrapper fails to start with "gitea-mcp-server not found":
|
||||||
|
|
||||||
|
1. Verify `gitea-mcp-server` is installed: `pip list | grep gitea-mcp`
|
||||||
|
2. Install it manually: `pip install gitea-mcp-server`
|
||||||
|
3. In Docker: Rebuild the image
|
||||||
|
|
||||||
|
### Tool Filtering Not Working
|
||||||
|
|
||||||
|
If tool filtering is not applied:
|
||||||
|
|
||||||
|
1. Check `.env` file syntax (no spaces around `=`)
|
||||||
|
2. Verify comma-separated list format
|
||||||
|
3. Check server logs for filter configuration
|
||||||
|
4. Query `POST /tools/list` to see filtered tools
|
||||||
|
|
||||||
|
## Security Considerations
|
||||||
|
|
||||||
|
- **Always use HTTPS** in production (configure reverse proxy)
|
||||||
|
- **Set AUTH_TOKEN** to secure the HTTP endpoint
|
||||||
|
- **Rotate tokens regularly** (both Gitea token and auth token)
|
||||||
|
- **Use secrets management** (not .env files) in production
|
||||||
|
- **Limit network exposure** (firewall, VPN, or private network)
|
||||||
|
- **Monitor access logs** for suspicious activity
|
||||||
|
|
||||||
|
## Contributing
|
||||||
|
|
||||||
|
Contributions are welcome! Please feel free to submit issues or pull requests.
|
||||||
|
|
||||||
## License
|
## License
|
||||||
|
|
||||||
MIT
|
MIT License - see LICENSE file for details
|
||||||
|
|
||||||
|
## Version
|
||||||
|
|
||||||
|
Current version: 0.1.0
|
||||||
|
|
||||||
|
## Author
|
||||||
|
|
||||||
|
Leo Miranda
|
||||||
|
|
||||||
|
## Links
|
||||||
|
|
||||||
|
- Repository: https://github.com/lmiranda/gitea-mcp-remote
|
||||||
|
- Issues: https://github.com/lmiranda/gitea-mcp-remote/issues
|
||||||
|
- Official Gitea MCP Server: https://github.com/modelcontextprotocol/servers/tree/main/src/gitea
|
||||||
|
- MCP Documentation: https://modelcontextprotocol.io
|
||||||
|
|||||||
43
docker-compose.yml
Normal file
43
docker-compose.yml
Normal file
@@ -0,0 +1,43 @@
|
|||||||
|
version: '3.8'
|
||||||
|
|
||||||
|
services:
|
||||||
|
gitea-mcp-wrapper:
|
||||||
|
build:
|
||||||
|
context: .
|
||||||
|
dockerfile: Dockerfile
|
||||||
|
image: gitea-mcp-wrapper:latest
|
||||||
|
container_name: gitea-mcp-wrapper
|
||||||
|
restart: unless-stopped
|
||||||
|
ports:
|
||||||
|
- "8000:8000"
|
||||||
|
environment:
|
||||||
|
# Gitea Configuration
|
||||||
|
- GITEA_URL=${GITEA_URL}
|
||||||
|
- GITEA_TOKEN=${GITEA_TOKEN}
|
||||||
|
- GITEA_OWNER=${GITEA_OWNER}
|
||||||
|
- GITEA_REPO=${GITEA_REPO}
|
||||||
|
|
||||||
|
# HTTP Server Configuration
|
||||||
|
- HTTP_HOST=0.0.0.0
|
||||||
|
- HTTP_PORT=8000
|
||||||
|
|
||||||
|
# Authentication (Optional)
|
||||||
|
- AUTH_TOKEN=${AUTH_TOKEN:-}
|
||||||
|
|
||||||
|
# Tool Filtering (Optional)
|
||||||
|
- ENABLED_TOOLS=${ENABLED_TOOLS:-}
|
||||||
|
- DISABLED_TOOLS=${DISABLED_TOOLS:-}
|
||||||
|
|
||||||
|
healthcheck:
|
||||||
|
test: ["CMD", "python", "-c", "import urllib.request; urllib.request.urlopen('http://localhost:8000/health').read()"]
|
||||||
|
interval: 30s
|
||||||
|
timeout: 3s
|
||||||
|
retries: 3
|
||||||
|
start_period: 5s
|
||||||
|
|
||||||
|
networks:
|
||||||
|
- gitea-mcp-network
|
||||||
|
|
||||||
|
networks:
|
||||||
|
gitea-mcp-network:
|
||||||
|
driver: bridge
|
||||||
@@ -4,15 +4,15 @@ build-backend = "setuptools.build_meta"
|
|||||||
|
|
||||||
[project]
|
[project]
|
||||||
name = "gitea-mcp-remote"
|
name = "gitea-mcp-remote"
|
||||||
version = "0.1.0"
|
version = "1.0.0"
|
||||||
description = "MCP server for Gitea API integration"
|
description = "HTTP transport wrapper for Gitea MCP server"
|
||||||
readme = "README.md"
|
readme = "README.md"
|
||||||
requires-python = ">=3.10"
|
requires-python = ">=3.10"
|
||||||
license = { text = "MIT" }
|
license = { text = "MIT" }
|
||||||
authors = [
|
authors = [
|
||||||
{ name = "Leo Miranda", email = "lmiranda@example.com" }
|
{ name = "Leo Miranda", email = "lmiranda@example.com" }
|
||||||
]
|
]
|
||||||
keywords = ["mcp", "gitea", "api", "server"]
|
keywords = ["mcp", "gitea", "api", "server", "http", "wrapper"]
|
||||||
classifiers = [
|
classifiers = [
|
||||||
"Development Status :: 3 - Alpha",
|
"Development Status :: 3 - Alpha",
|
||||||
"Intended Audience :: Developers",
|
"Intended Audience :: Developers",
|
||||||
@@ -24,19 +24,27 @@ classifiers = [
|
|||||||
]
|
]
|
||||||
|
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"mcp>=0.1.0",
|
"mcp>=0.9.0",
|
||||||
"httpx>=0.24.0",
|
"uvicorn>=0.27.0",
|
||||||
|
"pydantic>=2.0.0",
|
||||||
|
"pydantic-settings>=2.0.0",
|
||||||
"python-dotenv>=1.0.0",
|
"python-dotenv>=1.0.0",
|
||||||
|
"starlette>=0.36.0",
|
||||||
|
# gitea-mcp-server - installed separately (not on PyPI yet)
|
||||||
|
# See: https://gitea.hotserv.cloud/personal-projects/leo-claude-mktplace
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.optional-dependencies]
|
[project.optional-dependencies]
|
||||||
dev = [
|
dev = [
|
||||||
"pytest>=7.0.0",
|
"pytest>=7.0.0",
|
||||||
"pytest-asyncio>=0.21.0",
|
"pytest-asyncio>=0.21.0",
|
||||||
|
"pytest-cov>=4.0.0",
|
||||||
|
"httpx>=0.24.0",
|
||||||
|
"starlette>=0.36.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
[project.scripts]
|
[project.scripts]
|
||||||
gitea-mcp = "gitea_mcp.server:main"
|
gitea-http-wrapper = "gitea_http_wrapper.server:main"
|
||||||
|
|
||||||
[project.urls]
|
[project.urls]
|
||||||
Homepage = "https://github.com/lmiranda/gitea-mcp-remote"
|
Homepage = "https://github.com/lmiranda/gitea-mcp-remote"
|
||||||
@@ -47,7 +55,7 @@ where = ["src"]
|
|||||||
|
|
||||||
[tool.pytest.ini_options]
|
[tool.pytest.ini_options]
|
||||||
asyncio_mode = "auto"
|
asyncio_mode = "auto"
|
||||||
testpaths = ["tests"]
|
testpaths = ["src/gitea_http_wrapper/tests"]
|
||||||
python_files = ["test_*.py"]
|
python_files = ["test_*.py"]
|
||||||
python_classes = ["Test*"]
|
python_classes = ["Test*"]
|
||||||
python_functions = ["test_*"]
|
python_functions = ["test_*"]
|
||||||
|
|||||||
18
pytest.ini
Normal file
18
pytest.ini
Normal file
@@ -0,0 +1,18 @@
|
|||||||
|
[pytest]
|
||||||
|
testpaths = src/gitea_http_wrapper/tests
|
||||||
|
python_files = test_*.py
|
||||||
|
python_classes = Test*
|
||||||
|
python_functions = test_*
|
||||||
|
asyncio_mode = auto
|
||||||
|
|
||||||
|
# Coverage options
|
||||||
|
addopts =
|
||||||
|
--verbose
|
||||||
|
--strict-markers
|
||||||
|
--tb=short
|
||||||
|
|
||||||
|
# Markers for test categorization
|
||||||
|
markers =
|
||||||
|
unit: Unit tests (fast, no external dependencies)
|
||||||
|
integration: Integration tests (may require external services)
|
||||||
|
slow: Slow-running tests
|
||||||
11
requirements.txt
Normal file
11
requirements.txt
Normal file
@@ -0,0 +1,11 @@
|
|||||||
|
# HTTP Transport Wrapper Dependencies
|
||||||
|
mcp>=0.9.0
|
||||||
|
uvicorn>=0.27.0
|
||||||
|
starlette>=0.36.0
|
||||||
|
pydantic>=2.0.0
|
||||||
|
pydantic-settings>=2.0.0
|
||||||
|
python-dotenv>=1.0.0
|
||||||
|
|
||||||
|
# Official Gitea MCP Server (to be wrapped)
|
||||||
|
# Install separately - not on PyPI yet
|
||||||
|
# See: https://gitea.hotserv.cloud/personal-projects/leo-claude-mktplace
|
||||||
21
run_tests.sh
Executable file
21
run_tests.sh
Executable file
@@ -0,0 +1,21 @@
|
|||||||
|
#!/bin/bash
|
||||||
|
# Test runner script for gitea-mcp-remote
|
||||||
|
|
||||||
|
set -e
|
||||||
|
|
||||||
|
echo "Running gitea-mcp-remote test suite..."
|
||||||
|
echo
|
||||||
|
|
||||||
|
# Install dev dependencies if not already installed
|
||||||
|
if ! command -v pytest &> /dev/null; then
|
||||||
|
echo "Installing dev dependencies..."
|
||||||
|
pip install -e ".[dev]"
|
||||||
|
echo
|
||||||
|
fi
|
||||||
|
|
||||||
|
# Run tests with coverage
|
||||||
|
echo "Running tests with coverage..."
|
||||||
|
python -m pytest tests/ -v --cov=src/gitea_mcp --cov-report=term-missing --cov-report=html
|
||||||
|
|
||||||
|
echo
|
||||||
|
echo "Coverage report saved to htmlcov/index.html"
|
||||||
17
src/gitea_http_wrapper/__init__.py
Normal file
17
src/gitea_http_wrapper/__init__.py
Normal file
@@ -0,0 +1,17 @@
|
|||||||
|
"""
|
||||||
|
Gitea HTTP MCP Wrapper
|
||||||
|
|
||||||
|
This package provides an HTTP transport wrapper around the official Gitea MCP server.
|
||||||
|
It handles configuration loading, tool filtering, and HTTP authentication middleware.
|
||||||
|
|
||||||
|
Architecture:
|
||||||
|
- config/: Configuration loader module
|
||||||
|
- middleware/: HTTP authentication middleware
|
||||||
|
- filtering/: Tool filtering for Claude Desktop compatibility
|
||||||
|
- server.py: Main HTTP MCP server implementation
|
||||||
|
"""
|
||||||
|
|
||||||
|
from .server import GiteaMCPWrapper, create_app, main
|
||||||
|
|
||||||
|
__version__ = "0.1.0"
|
||||||
|
__all__ = ["__version__", "GiteaMCPWrapper", "create_app", "main"]
|
||||||
5
src/gitea_http_wrapper/config/__init__.py
Normal file
5
src/gitea_http_wrapper/config/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
"""Configuration loader module."""
|
||||||
|
|
||||||
|
from .settings import GiteaSettings, load_settings
|
||||||
|
|
||||||
|
__all__ = ["GiteaSettings", "load_settings"]
|
||||||
113
src/gitea_http_wrapper/config/settings.py
Normal file
113
src/gitea_http_wrapper/config/settings.py
Normal file
@@ -0,0 +1,113 @@
|
|||||||
|
"""Configuration settings for Gitea HTTP MCP wrapper."""
|
||||||
|
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Optional
|
||||||
|
|
||||||
|
from pydantic import Field, field_validator
|
||||||
|
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||||
|
|
||||||
|
|
||||||
|
class GiteaSettings(BaseSettings):
|
||||||
|
"""Configuration settings loaded from environment or .env file."""
|
||||||
|
|
||||||
|
model_config = SettingsConfigDict(
|
||||||
|
env_file=".env",
|
||||||
|
env_file_encoding="utf-8",
|
||||||
|
case_sensitive=False,
|
||||||
|
extra="ignore",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Gitea Configuration
|
||||||
|
gitea_url: str = Field(
|
||||||
|
...,
|
||||||
|
description="Gitea instance URL (e.g., https://git.example.com)",
|
||||||
|
)
|
||||||
|
gitea_token: str = Field(
|
||||||
|
...,
|
||||||
|
description="Gitea API token for authentication",
|
||||||
|
)
|
||||||
|
gitea_owner: str = Field(
|
||||||
|
...,
|
||||||
|
description="Default repository owner/organization",
|
||||||
|
)
|
||||||
|
gitea_repo: str = Field(
|
||||||
|
...,
|
||||||
|
description="Default repository name",
|
||||||
|
)
|
||||||
|
|
||||||
|
# HTTP Server Configuration
|
||||||
|
http_host: str = Field(
|
||||||
|
default="127.0.0.1",
|
||||||
|
description="HTTP server bind address",
|
||||||
|
)
|
||||||
|
http_port: int = Field(
|
||||||
|
default=8000,
|
||||||
|
ge=1,
|
||||||
|
le=65535,
|
||||||
|
description="HTTP server port",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Authentication Configuration
|
||||||
|
auth_token: Optional[str] = Field(
|
||||||
|
default=None,
|
||||||
|
description="Bearer token for HTTP authentication (optional)",
|
||||||
|
)
|
||||||
|
|
||||||
|
# Tool Filtering Configuration
|
||||||
|
enabled_tools: Optional[str] = Field(
|
||||||
|
default=None,
|
||||||
|
description="Comma-separated list of enabled tools (optional, enables all if not set)",
|
||||||
|
)
|
||||||
|
disabled_tools: Optional[str] = Field(
|
||||||
|
default=None,
|
||||||
|
description="Comma-separated list of disabled tools (optional)",
|
||||||
|
)
|
||||||
|
|
||||||
|
@field_validator("gitea_url")
|
||||||
|
@classmethod
|
||||||
|
def validate_gitea_url(cls, v: str) -> str:
|
||||||
|
"""Ensure Gitea URL is properly formatted."""
|
||||||
|
if not v.startswith(("http://", "https://")):
|
||||||
|
raise ValueError("gitea_url must start with http:// or https://")
|
||||||
|
return v.rstrip("/")
|
||||||
|
|
||||||
|
@property
|
||||||
|
def enabled_tools_list(self) -> Optional[list[str]]:
|
||||||
|
"""Parse enabled_tools into a list."""
|
||||||
|
if not self.enabled_tools:
|
||||||
|
return None
|
||||||
|
return [tool.strip() for tool in self.enabled_tools.split(",") if tool.strip()]
|
||||||
|
|
||||||
|
@property
|
||||||
|
def disabled_tools_list(self) -> Optional[list[str]]:
|
||||||
|
"""Parse disabled_tools into a list."""
|
||||||
|
if not self.disabled_tools:
|
||||||
|
return None
|
||||||
|
return [tool.strip() for tool in self.disabled_tools.split(",") if tool.strip()]
|
||||||
|
|
||||||
|
def get_gitea_mcp_env(self) -> dict[str, str]:
|
||||||
|
"""Get environment variables for the wrapped Gitea MCP server."""
|
||||||
|
return {
|
||||||
|
"GITEA_BASE_URL": self.gitea_url,
|
||||||
|
"GITEA_API_TOKEN": self.gitea_token,
|
||||||
|
"GITEA_DEFAULT_OWNER": self.gitea_owner,
|
||||||
|
"GITEA_DEFAULT_REPO": self.gitea_repo,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
def load_settings(env_file: Optional[Path] = None) -> GiteaSettings:
|
||||||
|
"""
|
||||||
|
Load settings from environment or .env file.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
env_file: Optional path to .env file. If not provided, searches for .env in current directory.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
GiteaSettings instance with loaded configuration.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValidationError: If required settings are missing or invalid.
|
||||||
|
"""
|
||||||
|
if env_file:
|
||||||
|
return GiteaSettings(_env_file=env_file)
|
||||||
|
return GiteaSettings()
|
||||||
5
src/gitea_http_wrapper/filtering/__init__.py
Normal file
5
src/gitea_http_wrapper/filtering/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
"""Tool filtering module for Claude Desktop compatibility."""
|
||||||
|
|
||||||
|
from .filter import ToolFilter
|
||||||
|
|
||||||
|
__all__ = ["ToolFilter"]
|
||||||
108
src/gitea_http_wrapper/filtering/filter.py
Normal file
108
src/gitea_http_wrapper/filtering/filter.py
Normal file
@@ -0,0 +1,108 @@
|
|||||||
|
"""Tool filtering for Claude Desktop compatibility."""
|
||||||
|
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
|
||||||
|
class ToolFilter:
|
||||||
|
"""
|
||||||
|
Filter MCP tools based on enabled/disabled lists.
|
||||||
|
|
||||||
|
This class handles tool filtering to ensure only compatible tools are exposed
|
||||||
|
to Claude Desktop, preventing crashes from unsupported tool schemas.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(
|
||||||
|
self,
|
||||||
|
enabled_tools: list[str] | None = None,
|
||||||
|
disabled_tools: list[str] | None = None,
|
||||||
|
):
|
||||||
|
"""
|
||||||
|
Initialize tool filter.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
enabled_tools: List of tool names to enable. If None, all tools are enabled.
|
||||||
|
disabled_tools: List of tool names to disable. Takes precedence over enabled_tools.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: If both enabled_tools and disabled_tools are specified.
|
||||||
|
"""
|
||||||
|
if enabled_tools is not None and disabled_tools is not None:
|
||||||
|
raise ValueError(
|
||||||
|
"Cannot specify both enabled_tools and disabled_tools. Choose one filtering mode."
|
||||||
|
)
|
||||||
|
|
||||||
|
self.enabled_tools = set(enabled_tools) if enabled_tools else None
|
||||||
|
self.disabled_tools = set(disabled_tools) if disabled_tools else None
|
||||||
|
|
||||||
|
def should_include_tool(self, tool_name: str) -> bool:
|
||||||
|
"""
|
||||||
|
Determine if a tool should be included based on filter rules.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
tool_name: Name of the tool to check.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
True if tool should be included, False otherwise.
|
||||||
|
"""
|
||||||
|
# If disabled list is specified, exclude disabled tools
|
||||||
|
if self.disabled_tools is not None:
|
||||||
|
return tool_name not in self.disabled_tools
|
||||||
|
|
||||||
|
# If enabled list is specified, only include enabled tools
|
||||||
|
if self.enabled_tools is not None:
|
||||||
|
return tool_name in self.enabled_tools
|
||||||
|
|
||||||
|
# If no filters specified, include all tools
|
||||||
|
return True
|
||||||
|
|
||||||
|
def filter_tools_list(self, tools: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||||||
|
"""
|
||||||
|
Filter a list of tool definitions.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
tools: List of tool definitions (dicts with at least a 'name' field).
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Filtered list of tool definitions.
|
||||||
|
"""
|
||||||
|
return [tool for tool in tools if self.should_include_tool(tool.get("name", ""))]
|
||||||
|
|
||||||
|
def filter_tools_response(self, response: dict[str, Any]) -> dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Filter tools from an MCP list_tools response.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
response: MCP response dict containing 'tools' list.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Filtered response with tools list updated.
|
||||||
|
"""
|
||||||
|
if "tools" in response and isinstance(response["tools"], list):
|
||||||
|
response = response.copy()
|
||||||
|
response["tools"] = self.filter_tools_list(response["tools"])
|
||||||
|
return response
|
||||||
|
|
||||||
|
def get_filter_stats(self) -> dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Get statistics about the filter configuration.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Dict containing filter mode and tool counts.
|
||||||
|
"""
|
||||||
|
if self.disabled_tools is not None:
|
||||||
|
return {
|
||||||
|
"mode": "blacklist",
|
||||||
|
"disabled_count": len(self.disabled_tools),
|
||||||
|
"disabled_tools": sorted(self.disabled_tools),
|
||||||
|
}
|
||||||
|
elif self.enabled_tools is not None:
|
||||||
|
return {
|
||||||
|
"mode": "whitelist",
|
||||||
|
"enabled_count": len(self.enabled_tools),
|
||||||
|
"enabled_tools": sorted(self.enabled_tools),
|
||||||
|
}
|
||||||
|
else:
|
||||||
|
return {
|
||||||
|
"mode": "passthrough",
|
||||||
|
"message": "All tools enabled",
|
||||||
|
}
|
||||||
5
src/gitea_http_wrapper/middleware/__init__.py
Normal file
5
src/gitea_http_wrapper/middleware/__init__.py
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
"""HTTP authentication middleware module."""
|
||||||
|
|
||||||
|
from .auth import BearerAuthMiddleware, HealthCheckBypassMiddleware
|
||||||
|
|
||||||
|
__all__ = ["BearerAuthMiddleware", "HealthCheckBypassMiddleware"]
|
||||||
144
src/gitea_http_wrapper/middleware/auth.py
Normal file
144
src/gitea_http_wrapper/middleware/auth.py
Normal file
@@ -0,0 +1,144 @@
|
|||||||
|
"""HTTP authentication middleware for MCP server."""
|
||||||
|
|
||||||
|
import logging
|
||||||
|
from typing import Awaitable, Callable
|
||||||
|
|
||||||
|
from starlette.middleware.base import BaseHTTPMiddleware
|
||||||
|
from starlette.requests import Request
|
||||||
|
from starlette.responses import JSONResponse, Response
|
||||||
|
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class BearerAuthMiddleware(BaseHTTPMiddleware):
|
||||||
|
"""
|
||||||
|
Middleware to enforce Bearer token authentication on HTTP requests.
|
||||||
|
|
||||||
|
This middleware validates the Authorization header for all requests.
|
||||||
|
If a token is configured, requests must include "Authorization: Bearer <token>".
|
||||||
|
If no token is configured, all requests are allowed (open access).
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, app, auth_token: str | None = None):
|
||||||
|
"""
|
||||||
|
Initialize authentication middleware.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
app: ASGI application to wrap.
|
||||||
|
auth_token: Optional Bearer token for authentication.
|
||||||
|
If None, authentication is disabled.
|
||||||
|
"""
|
||||||
|
super().__init__(app)
|
||||||
|
self.auth_token = auth_token
|
||||||
|
self.auth_enabled = auth_token is not None
|
||||||
|
|
||||||
|
if self.auth_enabled:
|
||||||
|
logger.info("Bearer authentication enabled")
|
||||||
|
else:
|
||||||
|
logger.warning("Bearer authentication disabled - server is open access")
|
||||||
|
|
||||||
|
async def dispatch(
|
||||||
|
self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
|
||||||
|
) -> Response:
|
||||||
|
"""
|
||||||
|
Process request and enforce authentication if enabled.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
request: Incoming HTTP request.
|
||||||
|
call_next: Next middleware or route handler.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Response from downstream handler or 401/403 error.
|
||||||
|
"""
|
||||||
|
# Skip authentication if disabled
|
||||||
|
if not self.auth_enabled:
|
||||||
|
return await call_next(request)
|
||||||
|
|
||||||
|
# Skip authentication if marked by HealthCheckBypassMiddleware
|
||||||
|
if getattr(request.state, "skip_auth", False):
|
||||||
|
return await call_next(request)
|
||||||
|
|
||||||
|
# Extract Authorization header
|
||||||
|
auth_header = request.headers.get("Authorization")
|
||||||
|
|
||||||
|
# Check if header is present
|
||||||
|
if not auth_header:
|
||||||
|
logger.warning(f"Missing Authorization header from {request.client.host}")
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=401,
|
||||||
|
content={
|
||||||
|
"error": "Unauthorized",
|
||||||
|
"message": "Missing Authorization header",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Check if header format is correct
|
||||||
|
if not auth_header.startswith("Bearer "):
|
||||||
|
logger.warning(f"Invalid Authorization format from {request.client.host}")
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=401,
|
||||||
|
content={
|
||||||
|
"error": "Unauthorized",
|
||||||
|
"message": "Authorization header must use Bearer scheme",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Extract token
|
||||||
|
provided_token = auth_header[7:] # Remove "Bearer " prefix
|
||||||
|
|
||||||
|
# Validate token
|
||||||
|
if provided_token != self.auth_token:
|
||||||
|
logger.warning(f"Invalid token from {request.client.host}")
|
||||||
|
return JSONResponse(
|
||||||
|
status_code=403,
|
||||||
|
content={
|
||||||
|
"error": "Forbidden",
|
||||||
|
"message": "Invalid authentication token",
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
# Token is valid, proceed to next handler
|
||||||
|
logger.debug(f"Authenticated request from {request.client.host}")
|
||||||
|
return await call_next(request)
|
||||||
|
|
||||||
|
|
||||||
|
class HealthCheckBypassMiddleware(BaseHTTPMiddleware):
|
||||||
|
"""
|
||||||
|
Middleware to bypass authentication for health check endpoints.
|
||||||
|
|
||||||
|
This allows monitoring systems to check server health without authentication.
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, app, health_check_paths: list[str] | None = None):
|
||||||
|
"""
|
||||||
|
Initialize health check bypass middleware.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
app: ASGI application to wrap.
|
||||||
|
health_check_paths: List of paths to bypass authentication.
|
||||||
|
Defaults to ["/health", "/healthz", "/ping"].
|
||||||
|
"""
|
||||||
|
super().__init__(app)
|
||||||
|
self.health_check_paths = health_check_paths or ["/health", "/healthz", "/ping"]
|
||||||
|
|
||||||
|
async def dispatch(
|
||||||
|
self, request: Request, call_next: Callable[[Request], Awaitable[Response]]
|
||||||
|
) -> Response:
|
||||||
|
"""
|
||||||
|
Process request and bypass authentication for health checks.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
request: Incoming HTTP request.
|
||||||
|
call_next: Next middleware or route handler.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Response from downstream handler.
|
||||||
|
"""
|
||||||
|
# Check if request is for a health check endpoint
|
||||||
|
if request.url.path in self.health_check_paths:
|
||||||
|
logger.debug(f"Bypassing auth for health check: {request.url.path}")
|
||||||
|
# Mark request to skip authentication in BearerAuthMiddleware
|
||||||
|
request.state.skip_auth = True
|
||||||
|
|
||||||
|
# Continue to next middleware
|
||||||
|
return await call_next(request)
|
||||||
309
src/gitea_http_wrapper/server.py
Normal file
309
src/gitea_http_wrapper/server.py
Normal file
@@ -0,0 +1,309 @@
|
|||||||
|
"""HTTP MCP server implementation wrapping Gitea MCP."""
|
||||||
|
|
||||||
|
import asyncio
|
||||||
|
import json
|
||||||
|
import logging
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
|
import uvicorn
|
||||||
|
from mcp.server import Server
|
||||||
|
from mcp.server.stdio import stdio_server
|
||||||
|
from starlette.applications import Starlette
|
||||||
|
from starlette.requests import Request
|
||||||
|
from starlette.responses import JSONResponse
|
||||||
|
from starlette.routing import Route
|
||||||
|
|
||||||
|
from gitea_http_wrapper.config import GiteaSettings, load_settings
|
||||||
|
from gitea_http_wrapper.filtering import ToolFilter
|
||||||
|
from gitea_http_wrapper.middleware import (
|
||||||
|
BearerAuthMiddleware,
|
||||||
|
HealthCheckBypassMiddleware,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Configure logging
|
||||||
|
logging.basicConfig(
|
||||||
|
level=logging.INFO,
|
||||||
|
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
|
||||||
|
)
|
||||||
|
logger = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
class GiteaMCPWrapper:
|
||||||
|
"""
|
||||||
|
HTTP wrapper around the official Gitea MCP server.
|
||||||
|
|
||||||
|
This class manages:
|
||||||
|
1. Starting the Gitea MCP server as a subprocess with stdio transport
|
||||||
|
2. Proxying HTTP requests to the MCP server
|
||||||
|
3. Filtering tools based on configuration
|
||||||
|
4. Handling responses and errors
|
||||||
|
"""
|
||||||
|
|
||||||
|
def __init__(self, settings: GiteaSettings):
|
||||||
|
"""
|
||||||
|
Initialize the MCP wrapper.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
settings: Configuration settings for Gitea and HTTP server.
|
||||||
|
"""
|
||||||
|
self.settings = settings
|
||||||
|
self.tool_filter = ToolFilter(
|
||||||
|
enabled_tools=settings.enabled_tools_list,
|
||||||
|
disabled_tools=settings.disabled_tools_list,
|
||||||
|
)
|
||||||
|
self.process = None
|
||||||
|
self.reader = None
|
||||||
|
self.writer = None
|
||||||
|
|
||||||
|
async def start_gitea_mcp(self) -> None:
|
||||||
|
"""
|
||||||
|
Start the Gitea MCP server as a subprocess.
|
||||||
|
|
||||||
|
The server runs with stdio transport, and we communicate via stdin/stdout.
|
||||||
|
"""
|
||||||
|
logger.info("Starting Gitea MCP server subprocess")
|
||||||
|
|
||||||
|
# Set environment variables for Gitea MCP
|
||||||
|
env = os.environ.copy()
|
||||||
|
env.update(self.settings.get_gitea_mcp_env())
|
||||||
|
|
||||||
|
# Start the process
|
||||||
|
# Note: This assumes gitea-mcp-server is installed and on PATH
|
||||||
|
# In production Docker, this should be guaranteed
|
||||||
|
try:
|
||||||
|
self.process = await asyncio.create_subprocess_exec(
|
||||||
|
"gitea-mcp-server",
|
||||||
|
stdin=asyncio.subprocess.PIPE,
|
||||||
|
stdout=asyncio.subprocess.PIPE,
|
||||||
|
stderr=asyncio.subprocess.PIPE,
|
||||||
|
env=env,
|
||||||
|
)
|
||||||
|
self.reader = self.process.stdout
|
||||||
|
self.writer = self.process.stdin
|
||||||
|
logger.info("Gitea MCP server started successfully")
|
||||||
|
except FileNotFoundError:
|
||||||
|
logger.error("gitea-mcp-server not found in PATH")
|
||||||
|
raise RuntimeError(
|
||||||
|
"gitea-mcp-server not found. Ensure it's installed: pip install gitea-mcp-server"
|
||||||
|
)
|
||||||
|
|
||||||
|
async def stop_gitea_mcp(self) -> None:
|
||||||
|
"""Stop the Gitea MCP server subprocess."""
|
||||||
|
if self.process:
|
||||||
|
logger.info("Stopping Gitea MCP server subprocess")
|
||||||
|
self.process.terminate()
|
||||||
|
await self.process.wait()
|
||||||
|
logger.info("Gitea MCP server stopped")
|
||||||
|
|
||||||
|
async def send_mcp_request(self, method: str, params: dict[str, Any]) -> dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Send a JSON-RPC request to the MCP server.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
method: MCP method name (e.g., "tools/list", "tools/call").
|
||||||
|
params: Method parameters.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
JSON-RPC response from MCP server.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
RuntimeError: If MCP server is not running or communication fails.
|
||||||
|
"""
|
||||||
|
if not self.writer or not self.reader:
|
||||||
|
raise RuntimeError("MCP server not started")
|
||||||
|
|
||||||
|
# Build JSON-RPC request
|
||||||
|
request = {
|
||||||
|
"jsonrpc": "2.0",
|
||||||
|
"id": 1,
|
||||||
|
"method": method,
|
||||||
|
"params": params,
|
||||||
|
}
|
||||||
|
|
||||||
|
# Send request
|
||||||
|
request_json = json.dumps(request) + "\n"
|
||||||
|
self.writer.write(request_json.encode())
|
||||||
|
await self.writer.drain()
|
||||||
|
|
||||||
|
# Read response
|
||||||
|
response_line = await self.reader.readline()
|
||||||
|
response = json.loads(response_line.decode())
|
||||||
|
|
||||||
|
# Check for JSON-RPC error
|
||||||
|
if "error" in response:
|
||||||
|
logger.error(f"MCP error: {response['error']}")
|
||||||
|
raise RuntimeError(f"MCP error: {response['error']}")
|
||||||
|
|
||||||
|
return response.get("result", {})
|
||||||
|
|
||||||
|
async def list_tools(self) -> dict[str, Any]:
|
||||||
|
"""
|
||||||
|
List available tools from MCP server with filtering applied.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Filtered tools list response.
|
||||||
|
"""
|
||||||
|
response = await self.send_mcp_request("tools/list", {})
|
||||||
|
filtered_response = self.tool_filter.filter_tools_response(response)
|
||||||
|
|
||||||
|
logger.info(
|
||||||
|
f"Listed {len(filtered_response.get('tools', []))} tools "
|
||||||
|
f"(filter: {self.tool_filter.get_filter_stats()['mode']})"
|
||||||
|
)
|
||||||
|
return filtered_response
|
||||||
|
|
||||||
|
async def call_tool(self, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
|
||||||
|
"""
|
||||||
|
Call a tool on the MCP server.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
tool_name: Name of tool to call.
|
||||||
|
arguments: Tool arguments.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Tool execution result.
|
||||||
|
|
||||||
|
Raises:
|
||||||
|
ValueError: If tool is filtered out.
|
||||||
|
"""
|
||||||
|
# Check if tool is allowed
|
||||||
|
if not self.tool_filter.should_include_tool(tool_name):
|
||||||
|
raise ValueError(f"Tool '{tool_name}' is not available (filtered)")
|
||||||
|
|
||||||
|
logger.info(f"Calling tool: {tool_name}")
|
||||||
|
result = await self.send_mcp_request(
|
||||||
|
"tools/call",
|
||||||
|
{"name": tool_name, "arguments": arguments},
|
||||||
|
)
|
||||||
|
return result
|
||||||
|
|
||||||
|
|
||||||
|
# Global wrapper instance
|
||||||
|
wrapper: GiteaMCPWrapper | None = None
|
||||||
|
|
||||||
|
|
||||||
|
async def health_check(request: Request) -> JSONResponse:
|
||||||
|
"""Health check endpoint."""
|
||||||
|
return JSONResponse({"status": "healthy"})
|
||||||
|
|
||||||
|
|
||||||
|
async def list_tools_endpoint(request: Request) -> JSONResponse:
|
||||||
|
"""List available tools."""
|
||||||
|
try:
|
||||||
|
tools = await wrapper.list_tools()
|
||||||
|
return JSONResponse(tools)
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("Error listing tools")
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": str(e)},
|
||||||
|
status_code=500,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def call_tool_endpoint(request: Request) -> JSONResponse:
|
||||||
|
"""Call a tool."""
|
||||||
|
try:
|
||||||
|
body = await request.json()
|
||||||
|
tool_name = body.get("name")
|
||||||
|
arguments = body.get("arguments", {})
|
||||||
|
|
||||||
|
if not tool_name:
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": "Missing 'name' field"},
|
||||||
|
status_code=400,
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await wrapper.call_tool(tool_name, arguments)
|
||||||
|
return JSONResponse(result)
|
||||||
|
except ValueError as e:
|
||||||
|
# Tool filtered
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": str(e)},
|
||||||
|
status_code=403,
|
||||||
|
)
|
||||||
|
except Exception as e:
|
||||||
|
logger.exception("Error calling tool")
|
||||||
|
return JSONResponse(
|
||||||
|
{"error": str(e)},
|
||||||
|
status_code=500,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
async def startup() -> None:
|
||||||
|
"""Application startup handler."""
|
||||||
|
global wrapper
|
||||||
|
settings = load_settings()
|
||||||
|
wrapper = GiteaMCPWrapper(settings)
|
||||||
|
await wrapper.start_gitea_mcp()
|
||||||
|
logger.info(f"HTTP MCP server starting on {settings.http_host}:{settings.http_port}")
|
||||||
|
|
||||||
|
|
||||||
|
async def shutdown() -> None:
|
||||||
|
"""Application shutdown handler."""
|
||||||
|
global wrapper
|
||||||
|
if wrapper:
|
||||||
|
await wrapper.stop_gitea_mcp()
|
||||||
|
|
||||||
|
|
||||||
|
# Define routes
|
||||||
|
routes = [
|
||||||
|
Route("/health", health_check, methods=["GET"]),
|
||||||
|
Route("/healthz", health_check, methods=["GET"]),
|
||||||
|
Route("/ping", health_check, methods=["GET"]),
|
||||||
|
Route("/tools/list", list_tools_endpoint, methods=["POST"]),
|
||||||
|
Route("/tools/call", call_tool_endpoint, methods=["POST"]),
|
||||||
|
]
|
||||||
|
|
||||||
|
# Create Starlette app
|
||||||
|
app = Starlette(
|
||||||
|
routes=routes,
|
||||||
|
on_startup=[startup],
|
||||||
|
on_shutdown=[shutdown],
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def create_app(settings: GiteaSettings | None = None) -> Starlette:
|
||||||
|
"""
|
||||||
|
Create and configure the Starlette application.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
settings: Optional settings override for testing.
|
||||||
|
|
||||||
|
Returns:
|
||||||
|
Configured Starlette application.
|
||||||
|
"""
|
||||||
|
if settings is None:
|
||||||
|
settings = load_settings()
|
||||||
|
|
||||||
|
# Add middleware
|
||||||
|
app.add_middleware(HealthCheckBypassMiddleware)
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token=settings.auth_token)
|
||||||
|
|
||||||
|
return app
|
||||||
|
|
||||||
|
|
||||||
|
def main() -> None:
|
||||||
|
"""Main entry point for the HTTP MCP server."""
|
||||||
|
settings = load_settings()
|
||||||
|
|
||||||
|
# Log filter configuration
|
||||||
|
filter_stats = ToolFilter(
|
||||||
|
enabled_tools=settings.enabled_tools_list,
|
||||||
|
disabled_tools=settings.disabled_tools_list,
|
||||||
|
).get_filter_stats()
|
||||||
|
logger.info(f"Tool filtering: {filter_stats}")
|
||||||
|
|
||||||
|
# Run server
|
||||||
|
uvicorn.run(
|
||||||
|
"gitea_http_wrapper.server:app",
|
||||||
|
host=settings.http_host,
|
||||||
|
port=settings.http_port,
|
||||||
|
log_level="info",
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
main()
|
||||||
9
src/gitea_http_wrapper/tests/__init__.py
Normal file
9
src/gitea_http_wrapper/tests/__init__.py
Normal file
@@ -0,0 +1,9 @@
|
|||||||
|
"""Test suite for HTTP wrapper functionality."""
|
||||||
|
|
||||||
|
# This package contains tests for:
|
||||||
|
# - config: Configuration loader and validation
|
||||||
|
# - filtering: Tool filtering for Claude Desktop compatibility
|
||||||
|
# - middleware: HTTP authentication middleware
|
||||||
|
# - server: Core HTTP MCP server (integration tests would go here)
|
||||||
|
|
||||||
|
__all__ = []
|
||||||
59
src/gitea_http_wrapper/tests/conftest.py
Normal file
59
src/gitea_http_wrapper/tests/conftest.py
Normal file
@@ -0,0 +1,59 @@
|
|||||||
|
"""Pytest configuration and shared fixtures for test suite."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def sample_gitea_config():
|
||||||
|
"""Provide sample Gitea configuration for tests."""
|
||||||
|
return {
|
||||||
|
"gitea_url": "https://gitea.test.com",
|
||||||
|
"gitea_token": "test_token_123",
|
||||||
|
"gitea_owner": "test_owner",
|
||||||
|
"gitea_repo": "test_repo",
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def sample_tools_list():
|
||||||
|
"""Provide sample MCP tools list for testing."""
|
||||||
|
return [
|
||||||
|
{
|
||||||
|
"name": "list_issues",
|
||||||
|
"description": "List issues in repository",
|
||||||
|
"inputSchema": {
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"state": {"type": "string", "enum": ["open", "closed", "all"]},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"name": "create_issue",
|
||||||
|
"description": "Create a new issue",
|
||||||
|
"inputSchema": {
|
||||||
|
"type": "object",
|
||||||
|
"properties": {
|
||||||
|
"title": {"type": "string"},
|
||||||
|
"body": {"type": "string"},
|
||||||
|
},
|
||||||
|
"required": ["title"],
|
||||||
|
},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
"name": "list_labels",
|
||||||
|
"description": "List labels in repository",
|
||||||
|
"inputSchema": {"type": "object", "properties": {}},
|
||||||
|
},
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def sample_mcp_response(sample_tools_list):
|
||||||
|
"""Provide sample MCP list_tools response."""
|
||||||
|
return {
|
||||||
|
"tools": sample_tools_list,
|
||||||
|
"meta": {
|
||||||
|
"version": "1.0",
|
||||||
|
},
|
||||||
|
}
|
||||||
211
src/gitea_http_wrapper/tests/test_config.py
Normal file
211
src/gitea_http_wrapper/tests/test_config.py
Normal file
@@ -0,0 +1,211 @@
|
|||||||
|
"""Tests for configuration loader module."""
|
||||||
|
|
||||||
|
import os
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from pydantic import ValidationError
|
||||||
|
|
||||||
|
from gitea_http_wrapper.config import GiteaSettings, load_settings
|
||||||
|
|
||||||
|
|
||||||
|
class TestGiteaSettings:
|
||||||
|
"""Test GiteaSettings configuration class."""
|
||||||
|
|
||||||
|
def test_required_fields(self):
|
||||||
|
"""Test that required fields are enforced."""
|
||||||
|
with pytest.raises(ValidationError) as exc_info:
|
||||||
|
GiteaSettings()
|
||||||
|
|
||||||
|
errors = exc_info.value.errors()
|
||||||
|
# Note: gitea_repo is optional (for PMO mode)
|
||||||
|
required_fields = {"gitea_url", "gitea_token", "gitea_owner"}
|
||||||
|
error_fields = {error["loc"][0] for error in errors}
|
||||||
|
assert required_fields.issubset(error_fields)
|
||||||
|
|
||||||
|
def test_valid_configuration(self):
|
||||||
|
"""Test valid configuration creation."""
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="test_token",
|
||||||
|
gitea_owner="test_owner",
|
||||||
|
gitea_repo="test_repo",
|
||||||
|
)
|
||||||
|
|
||||||
|
assert settings.gitea_url == "https://gitea.example.com"
|
||||||
|
assert settings.gitea_token == "test_token"
|
||||||
|
assert settings.gitea_owner == "test_owner"
|
||||||
|
assert settings.gitea_repo == "test_repo"
|
||||||
|
assert settings.http_host == "127.0.0.1"
|
||||||
|
assert settings.http_port == 8000
|
||||||
|
assert settings.auth_token is None
|
||||||
|
|
||||||
|
def test_gitea_url_validation(self):
|
||||||
|
"""Test Gitea URL validation."""
|
||||||
|
# Valid URLs
|
||||||
|
valid_urls = [
|
||||||
|
"http://gitea.local",
|
||||||
|
"https://gitea.example.com",
|
||||||
|
"http://192.168.1.1:3000",
|
||||||
|
]
|
||||||
|
|
||||||
|
for url in valid_urls:
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url=url,
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
)
|
||||||
|
assert settings.gitea_url == url.rstrip("/")
|
||||||
|
|
||||||
|
# Invalid URL (no protocol)
|
||||||
|
with pytest.raises(ValidationError) as exc_info:
|
||||||
|
GiteaSettings(
|
||||||
|
gitea_url="gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
)
|
||||||
|
assert "must start with http://" in str(exc_info.value)
|
||||||
|
|
||||||
|
def test_gitea_url_trailing_slash_removed(self):
|
||||||
|
"""Test that trailing slashes are removed from Gitea URL."""
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com/",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
)
|
||||||
|
assert settings.gitea_url == "https://gitea.example.com"
|
||||||
|
|
||||||
|
def test_http_port_validation(self):
|
||||||
|
"""Test HTTP port validation."""
|
||||||
|
# Valid port
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
http_port=9000,
|
||||||
|
)
|
||||||
|
assert settings.http_port == 9000
|
||||||
|
|
||||||
|
# Invalid port (too high)
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
http_port=70000,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Invalid port (too low)
|
||||||
|
with pytest.raises(ValidationError):
|
||||||
|
GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
http_port=0,
|
||||||
|
)
|
||||||
|
|
||||||
|
def test_enabled_tools_list_parsing(self):
|
||||||
|
"""Test enabled_tools string parsing to list."""
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
enabled_tools="tool1,tool2,tool3",
|
||||||
|
)
|
||||||
|
assert settings.enabled_tools_list == ["tool1", "tool2", "tool3"]
|
||||||
|
|
||||||
|
# Test with spaces
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
enabled_tools="tool1, tool2 , tool3",
|
||||||
|
)
|
||||||
|
assert settings.enabled_tools_list == ["tool1", "tool2", "tool3"]
|
||||||
|
|
||||||
|
# Test empty string
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
enabled_tools="",
|
||||||
|
)
|
||||||
|
assert settings.enabled_tools_list is None
|
||||||
|
|
||||||
|
def test_disabled_tools_list_parsing(self):
|
||||||
|
"""Test disabled_tools string parsing to list."""
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="token",
|
||||||
|
gitea_owner="owner",
|
||||||
|
gitea_repo="repo",
|
||||||
|
disabled_tools="tool1,tool2",
|
||||||
|
)
|
||||||
|
assert settings.disabled_tools_list == ["tool1", "tool2"]
|
||||||
|
|
||||||
|
def test_get_gitea_mcp_env(self):
|
||||||
|
"""Test environment variable generation for wrapped MCP server."""
|
||||||
|
settings = GiteaSettings(
|
||||||
|
gitea_url="https://gitea.example.com",
|
||||||
|
gitea_token="test_token",
|
||||||
|
gitea_owner="test_owner",
|
||||||
|
gitea_repo="test_repo",
|
||||||
|
)
|
||||||
|
|
||||||
|
env = settings.get_gitea_mcp_env()
|
||||||
|
|
||||||
|
assert env["GITEA_BASE_URL"] == "https://gitea.example.com"
|
||||||
|
assert env["GITEA_API_TOKEN"] == "test_token"
|
||||||
|
assert env["GITEA_DEFAULT_OWNER"] == "test_owner"
|
||||||
|
assert env["GITEA_DEFAULT_REPO"] == "test_repo"
|
||||||
|
|
||||||
|
|
||||||
|
class TestLoadSettings:
|
||||||
|
"""Test load_settings factory function."""
|
||||||
|
|
||||||
|
def test_load_from_env_file(self, tmp_path):
|
||||||
|
"""Test loading settings from a .env file."""
|
||||||
|
env_file = tmp_path / ".env"
|
||||||
|
env_file.write_text(
|
||||||
|
"""
|
||||||
|
GITEA_URL=https://gitea.test.com
|
||||||
|
GITEA_TOKEN=test_token_123
|
||||||
|
GITEA_OWNER=test_owner
|
||||||
|
GITEA_REPO=test_repo
|
||||||
|
HTTP_PORT=9000
|
||||||
|
"""
|
||||||
|
)
|
||||||
|
|
||||||
|
settings = load_settings(env_file)
|
||||||
|
|
||||||
|
assert settings.gitea_url == "https://gitea.test.com"
|
||||||
|
assert settings.gitea_token == "test_token_123"
|
||||||
|
assert settings.gitea_owner == "test_owner"
|
||||||
|
assert settings.gitea_repo == "test_repo"
|
||||||
|
assert settings.http_port == 9000
|
||||||
|
|
||||||
|
def test_load_from_environment(self, monkeypatch):
|
||||||
|
"""Test loading settings from environment variables."""
|
||||||
|
monkeypatch.setenv("GITEA_URL", "https://env.gitea.com")
|
||||||
|
monkeypatch.setenv("GITEA_TOKEN", "env_token")
|
||||||
|
monkeypatch.setenv("GITEA_OWNER", "env_owner")
|
||||||
|
monkeypatch.setenv("GITEA_REPO", "env_repo")
|
||||||
|
monkeypatch.setenv("HTTP_PORT", "8080")
|
||||||
|
|
||||||
|
# Mock _env_file to prevent loading actual .env
|
||||||
|
settings = GiteaSettings()
|
||||||
|
|
||||||
|
assert settings.gitea_url == "https://env.gitea.com"
|
||||||
|
assert settings.gitea_token == "env_token"
|
||||||
|
assert settings.gitea_owner == "env_owner"
|
||||||
|
assert settings.gitea_repo == "env_repo"
|
||||||
|
assert settings.http_port == 8080
|
||||||
143
src/gitea_http_wrapper/tests/test_filtering.py
Normal file
143
src/gitea_http_wrapper/tests/test_filtering.py
Normal file
@@ -0,0 +1,143 @@
|
|||||||
|
"""Tests for tool filtering module."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from gitea_http_wrapper.filtering import ToolFilter
|
||||||
|
|
||||||
|
|
||||||
|
class TestToolFilter:
|
||||||
|
"""Test ToolFilter class."""
|
||||||
|
|
||||||
|
def test_init_with_both_lists_raises(self):
|
||||||
|
"""Test that specifying both enabled and disabled lists raises error."""
|
||||||
|
with pytest.raises(ValueError) as exc_info:
|
||||||
|
ToolFilter(enabled_tools=["tool1"], disabled_tools=["tool2"])
|
||||||
|
|
||||||
|
assert "Cannot specify both" in str(exc_info.value)
|
||||||
|
|
||||||
|
def test_passthrough_mode(self):
|
||||||
|
"""Test passthrough mode (no filtering)."""
|
||||||
|
filter = ToolFilter()
|
||||||
|
|
||||||
|
assert filter.should_include_tool("any_tool")
|
||||||
|
assert filter.should_include_tool("another_tool")
|
||||||
|
|
||||||
|
stats = filter.get_filter_stats()
|
||||||
|
assert stats["mode"] == "passthrough"
|
||||||
|
|
||||||
|
def test_whitelist_mode(self):
|
||||||
|
"""Test whitelist mode (enabled_tools)."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1", "tool2"])
|
||||||
|
|
||||||
|
assert filter.should_include_tool("tool1")
|
||||||
|
assert filter.should_include_tool("tool2")
|
||||||
|
assert not filter.should_include_tool("tool3")
|
||||||
|
assert not filter.should_include_tool("tool4")
|
||||||
|
|
||||||
|
stats = filter.get_filter_stats()
|
||||||
|
assert stats["mode"] == "whitelist"
|
||||||
|
assert stats["enabled_count"] == 2
|
||||||
|
assert "tool1" in stats["enabled_tools"]
|
||||||
|
assert "tool2" in stats["enabled_tools"]
|
||||||
|
|
||||||
|
def test_blacklist_mode(self):
|
||||||
|
"""Test blacklist mode (disabled_tools)."""
|
||||||
|
filter = ToolFilter(disabled_tools=["tool1", "tool2"])
|
||||||
|
|
||||||
|
assert not filter.should_include_tool("tool1")
|
||||||
|
assert not filter.should_include_tool("tool2")
|
||||||
|
assert filter.should_include_tool("tool3")
|
||||||
|
assert filter.should_include_tool("tool4")
|
||||||
|
|
||||||
|
stats = filter.get_filter_stats()
|
||||||
|
assert stats["mode"] == "blacklist"
|
||||||
|
assert stats["disabled_count"] == 2
|
||||||
|
assert "tool1" in stats["disabled_tools"]
|
||||||
|
assert "tool2" in stats["disabled_tools"]
|
||||||
|
|
||||||
|
def test_filter_tools_list(self):
|
||||||
|
"""Test filtering a list of tool definitions."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1", "tool3"])
|
||||||
|
|
||||||
|
tools = [
|
||||||
|
{"name": "tool1", "description": "First tool"},
|
||||||
|
{"name": "tool2", "description": "Second tool"},
|
||||||
|
{"name": "tool3", "description": "Third tool"},
|
||||||
|
{"name": "tool4", "description": "Fourth tool"},
|
||||||
|
]
|
||||||
|
|
||||||
|
filtered = filter.filter_tools_list(tools)
|
||||||
|
|
||||||
|
assert len(filtered) == 2
|
||||||
|
assert filtered[0]["name"] == "tool1"
|
||||||
|
assert filtered[1]["name"] == "tool3"
|
||||||
|
|
||||||
|
def test_filter_tools_response(self):
|
||||||
|
"""Test filtering an MCP list_tools response."""
|
||||||
|
filter = ToolFilter(disabled_tools=["tool2"])
|
||||||
|
|
||||||
|
response = {
|
||||||
|
"tools": [
|
||||||
|
{"name": "tool1", "description": "First tool"},
|
||||||
|
{"name": "tool2", "description": "Second tool"},
|
||||||
|
{"name": "tool3", "description": "Third tool"},
|
||||||
|
],
|
||||||
|
"other_data": "preserved",
|
||||||
|
}
|
||||||
|
|
||||||
|
filtered = filter.filter_tools_response(response)
|
||||||
|
|
||||||
|
assert len(filtered["tools"]) == 2
|
||||||
|
assert filtered["tools"][0]["name"] == "tool1"
|
||||||
|
assert filtered["tools"][1]["name"] == "tool3"
|
||||||
|
assert filtered["other_data"] == "preserved"
|
||||||
|
|
||||||
|
def test_filter_tools_response_no_tools_key(self):
|
||||||
|
"""Test filtering response without 'tools' key."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1"])
|
||||||
|
|
||||||
|
response = {"other_data": "value"}
|
||||||
|
filtered = filter.filter_tools_response(response)
|
||||||
|
|
||||||
|
assert filtered == response
|
||||||
|
|
||||||
|
def test_filter_tools_response_immutable(self):
|
||||||
|
"""Test that original response is not mutated."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1"])
|
||||||
|
|
||||||
|
original = {
|
||||||
|
"tools": [
|
||||||
|
{"name": "tool1"},
|
||||||
|
{"name": "tool2"},
|
||||||
|
]
|
||||||
|
}
|
||||||
|
|
||||||
|
filtered = filter.filter_tools_response(original)
|
||||||
|
|
||||||
|
# Original should still have 2 tools
|
||||||
|
assert len(original["tools"]) == 2
|
||||||
|
# Filtered should have 1 tool
|
||||||
|
assert len(filtered["tools"]) == 1
|
||||||
|
|
||||||
|
def test_empty_tool_list(self):
|
||||||
|
"""Test filtering empty tool list."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1"])
|
||||||
|
|
||||||
|
result = filter.filter_tools_list([])
|
||||||
|
assert result == []
|
||||||
|
|
||||||
|
def test_tool_with_no_name(self):
|
||||||
|
"""Test handling tool without name field."""
|
||||||
|
filter = ToolFilter(enabled_tools=["tool1"])
|
||||||
|
|
||||||
|
tools = [
|
||||||
|
{"name": "tool1"},
|
||||||
|
{"description": "No name"},
|
||||||
|
{"name": "tool2"},
|
||||||
|
]
|
||||||
|
|
||||||
|
filtered = filter.filter_tools_list(tools)
|
||||||
|
|
||||||
|
# Only tool1 should match, tool without name is excluded
|
||||||
|
assert len(filtered) == 1
|
||||||
|
assert filtered[0]["name"] == "tool1"
|
||||||
162
src/gitea_http_wrapper/tests/test_middleware.py
Normal file
162
src/gitea_http_wrapper/tests/test_middleware.py
Normal file
@@ -0,0 +1,162 @@
|
|||||||
|
"""Tests for HTTP authentication middleware."""
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
from starlette.applications import Starlette
|
||||||
|
from starlette.responses import JSONResponse
|
||||||
|
from starlette.routing import Route
|
||||||
|
from starlette.testclient import TestClient
|
||||||
|
|
||||||
|
from gitea_http_wrapper.middleware import (
|
||||||
|
BearerAuthMiddleware,
|
||||||
|
HealthCheckBypassMiddleware,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
# Test application endpoint
|
||||||
|
async def test_endpoint(request):
|
||||||
|
return JSONResponse({"message": "success"})
|
||||||
|
|
||||||
|
|
||||||
|
class TestBearerAuthMiddleware:
|
||||||
|
"""Test BearerAuthMiddleware."""
|
||||||
|
|
||||||
|
def test_no_auth_configured(self):
|
||||||
|
"""Test that requests pass through when no auth token is configured."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token=None)
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
response = client.get("/test")
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json()["message"] == "success"
|
||||||
|
|
||||||
|
def test_auth_configured_valid_token(self):
|
||||||
|
"""Test successful authentication with valid token."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
response = client.get("/test", headers={"Authorization": "Bearer secret_token"})
|
||||||
|
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json()["message"] == "success"
|
||||||
|
|
||||||
|
def test_auth_configured_missing_header(self):
|
||||||
|
"""Test rejection when Authorization header is missing."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
response = client.get("/test")
|
||||||
|
|
||||||
|
assert response.status_code == 401
|
||||||
|
assert "Missing Authorization header" in response.json()["message"]
|
||||||
|
|
||||||
|
def test_auth_configured_invalid_format(self):
|
||||||
|
"""Test rejection when Authorization header has wrong format."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
# Test with wrong scheme
|
||||||
|
response = client.get("/test", headers={"Authorization": "Basic secret_token"})
|
||||||
|
assert response.status_code == 401
|
||||||
|
assert "Bearer scheme" in response.json()["message"]
|
||||||
|
|
||||||
|
# Test with no scheme
|
||||||
|
response = client.get("/test", headers={"Authorization": "secret_token"})
|
||||||
|
assert response.status_code == 401
|
||||||
|
|
||||||
|
def test_auth_configured_invalid_token(self):
|
||||||
|
"""Test rejection when token is invalid."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
response = client.get("/test", headers={"Authorization": "Bearer wrong_token"})
|
||||||
|
|
||||||
|
assert response.status_code == 403
|
||||||
|
assert "Invalid authentication token" in response.json()["message"]
|
||||||
|
|
||||||
|
def test_auth_case_sensitive_token(self):
|
||||||
|
"""Test that token comparison is case-sensitive."""
|
||||||
|
app = Starlette(routes=[Route("/test", test_endpoint)])
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="Secret_Token")
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
# Correct case
|
||||||
|
response = client.get("/test", headers={"Authorization": "Bearer Secret_Token"})
|
||||||
|
assert response.status_code == 200
|
||||||
|
|
||||||
|
# Wrong case
|
||||||
|
response = client.get("/test", headers={"Authorization": "Bearer secret_token"})
|
||||||
|
assert response.status_code == 403
|
||||||
|
|
||||||
|
|
||||||
|
class TestHealthCheckBypassMiddleware:
|
||||||
|
"""Test HealthCheckBypassMiddleware."""
|
||||||
|
|
||||||
|
def test_default_health_check_paths(self):
|
||||||
|
"""Test that default health check paths bypass auth."""
|
||||||
|
app = Starlette(
|
||||||
|
routes=[
|
||||||
|
Route("/health", test_endpoint),
|
||||||
|
Route("/healthz", test_endpoint),
|
||||||
|
Route("/ping", test_endpoint),
|
||||||
|
Route("/test", test_endpoint),
|
||||||
|
]
|
||||||
|
)
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
app.add_middleware(HealthCheckBypassMiddleware)
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
# Health checks should work without auth
|
||||||
|
assert client.get("/health").status_code == 200
|
||||||
|
assert client.get("/healthz").status_code == 200
|
||||||
|
assert client.get("/ping").status_code == 200
|
||||||
|
|
||||||
|
# Regular endpoint should require auth
|
||||||
|
assert client.get("/test").status_code == 401
|
||||||
|
|
||||||
|
def test_custom_health_check_paths(self):
|
||||||
|
"""Test custom health check paths."""
|
||||||
|
app = Starlette(
|
||||||
|
routes=[
|
||||||
|
Route("/custom-health", test_endpoint),
|
||||||
|
Route("/test", test_endpoint),
|
||||||
|
]
|
||||||
|
)
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
app.add_middleware(
|
||||||
|
HealthCheckBypassMiddleware,
|
||||||
|
health_check_paths=["/custom-health"],
|
||||||
|
)
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
|
||||||
|
# Custom health check should work without auth
|
||||||
|
assert client.get("/custom-health").status_code == 200
|
||||||
|
|
||||||
|
# Regular endpoint should require auth
|
||||||
|
assert client.get("/test").status_code == 401
|
||||||
|
|
||||||
|
def test_middleware_order(self):
|
||||||
|
"""Test that middleware order is correct."""
|
||||||
|
# HealthCheckBypass should be added BEFORE BearerAuth
|
||||||
|
# so it can bypass the auth check
|
||||||
|
|
||||||
|
app = Starlette(routes=[Route("/health", test_endpoint)])
|
||||||
|
|
||||||
|
# Correct order: HealthCheck bypass first, then Auth
|
||||||
|
app.add_middleware(BearerAuthMiddleware, auth_token="secret_token")
|
||||||
|
app.add_middleware(HealthCheckBypassMiddleware)
|
||||||
|
|
||||||
|
client = TestClient(app)
|
||||||
|
response = client.get("/health")
|
||||||
|
|
||||||
|
# Should succeed without auth
|
||||||
|
assert response.status_code == 200
|
||||||
@@ -1,3 +0,0 @@
|
|||||||
"""Gitea MCP Server - MCP server for Gitea API integration."""
|
|
||||||
|
|
||||||
__version__ = "0.1.0"
|
|
||||||
@@ -1,51 +0,0 @@
|
|||||||
"""Authentication and configuration management for Gitea MCP server."""
|
|
||||||
|
|
||||||
import os
|
|
||||||
from typing import Optional
|
|
||||||
from dotenv import load_dotenv
|
|
||||||
|
|
||||||
|
|
||||||
class AuthConfig:
|
|
||||||
"""Manages authentication configuration for Gitea API."""
|
|
||||||
|
|
||||||
def __init__(self):
|
|
||||||
"""Initialize authentication configuration from environment variables."""
|
|
||||||
load_dotenv()
|
|
||||||
|
|
||||||
self.api_url: Optional[str] = os.getenv("GITEA_API_URL")
|
|
||||||
self.api_token: Optional[str] = os.getenv("GITEA_API_TOKEN")
|
|
||||||
|
|
||||||
self._validate()
|
|
||||||
|
|
||||||
def _validate(self) -> None:
|
|
||||||
"""Validate that required configuration is present.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
ValueError: If required environment variables are missing.
|
|
||||||
"""
|
|
||||||
if not self.api_url:
|
|
||||||
raise ValueError(
|
|
||||||
"GITEA_API_URL environment variable is required. "
|
|
||||||
"Please set it in your .env file or environment."
|
|
||||||
)
|
|
||||||
|
|
||||||
if not self.api_token:
|
|
||||||
raise ValueError(
|
|
||||||
"GITEA_API_TOKEN environment variable is required. "
|
|
||||||
"Please set it in your .env file or environment."
|
|
||||||
)
|
|
||||||
|
|
||||||
# Remove trailing slash from URL if present
|
|
||||||
if self.api_url.endswith("/"):
|
|
||||||
self.api_url = self.api_url[:-1]
|
|
||||||
|
|
||||||
def get_auth_headers(self) -> dict[str, str]:
|
|
||||||
"""Get authentication headers for API requests.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
dict: HTTP headers with authorization token.
|
|
||||||
"""
|
|
||||||
return {
|
|
||||||
"Authorization": f"token {self.api_token}",
|
|
||||||
"Content-Type": "application/json",
|
|
||||||
}
|
|
||||||
@@ -1,216 +0,0 @@
|
|||||||
"""HTTP client for Gitea API."""
|
|
||||||
|
|
||||||
import httpx
|
|
||||||
from typing import Any, Optional
|
|
||||||
from .auth import AuthConfig
|
|
||||||
|
|
||||||
|
|
||||||
class GiteaClientError(Exception):
|
|
||||||
"""Base exception for Gitea client errors."""
|
|
||||||
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class GiteaAuthError(GiteaClientError):
|
|
||||||
"""Authentication error."""
|
|
||||||
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class GiteaNotFoundError(GiteaClientError):
|
|
||||||
"""Resource not found error."""
|
|
||||||
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class GiteaServerError(GiteaClientError):
|
|
||||||
"""Server error."""
|
|
||||||
|
|
||||||
pass
|
|
||||||
|
|
||||||
|
|
||||||
class GiteaClient:
|
|
||||||
"""Async HTTP client for Gitea API."""
|
|
||||||
|
|
||||||
def __init__(self, config: AuthConfig, timeout: float = 30.0):
|
|
||||||
"""Initialize Gitea API client.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
config: Authentication configuration.
|
|
||||||
timeout: Request timeout in seconds (default: 30.0).
|
|
||||||
"""
|
|
||||||
self.config = config
|
|
||||||
self.timeout = timeout
|
|
||||||
self._client: Optional[httpx.AsyncClient] = None
|
|
||||||
|
|
||||||
async def __aenter__(self):
|
|
||||||
"""Async context manager entry."""
|
|
||||||
self._client = httpx.AsyncClient(
|
|
||||||
base_url=self.config.api_url,
|
|
||||||
headers=self.config.get_auth_headers(),
|
|
||||||
timeout=self.timeout,
|
|
||||||
)
|
|
||||||
return self
|
|
||||||
|
|
||||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
|
||||||
"""Async context manager exit."""
|
|
||||||
if self._client:
|
|
||||||
await self._client.aclose()
|
|
||||||
|
|
||||||
def _handle_error(self, response: httpx.Response) -> None:
|
|
||||||
"""Handle HTTP error responses.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
response: HTTP response object.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
GiteaAuthError: For 401/403 errors.
|
|
||||||
GiteaNotFoundError: For 404 errors.
|
|
||||||
GiteaServerError: For 500+ errors.
|
|
||||||
GiteaClientError: For other errors.
|
|
||||||
"""
|
|
||||||
status = response.status_code
|
|
||||||
|
|
||||||
if status == 401:
|
|
||||||
raise GiteaAuthError(
|
|
||||||
"Authentication failed. Please check your GITEA_API_TOKEN."
|
|
||||||
)
|
|
||||||
elif status == 403:
|
|
||||||
raise GiteaAuthError(
|
|
||||||
"Access forbidden. Your API token may not have required permissions."
|
|
||||||
)
|
|
||||||
elif status == 404:
|
|
||||||
raise GiteaNotFoundError(
|
|
||||||
f"Resource not found: {response.request.url}"
|
|
||||||
)
|
|
||||||
elif status >= 500:
|
|
||||||
raise GiteaServerError(
|
|
||||||
f"Gitea server error (HTTP {status}): {response.text}"
|
|
||||||
)
|
|
||||||
else:
|
|
||||||
raise GiteaClientError(
|
|
||||||
f"API request failed (HTTP {status}): {response.text}"
|
|
||||||
)
|
|
||||||
|
|
||||||
async def get(self, path: str, **kwargs) -> dict[str, Any]:
|
|
||||||
"""Make GET request to Gitea API.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
path: API endpoint path (e.g., "/api/v1/repos/owner/repo").
|
|
||||||
**kwargs: Additional arguments for httpx request.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
dict: JSON response data.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
GiteaClientError: If request fails.
|
|
||||||
"""
|
|
||||||
if not self._client:
|
|
||||||
raise GiteaClientError(
|
|
||||||
"Client not initialized. Use 'async with' context manager."
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
response = await self._client.get(path, **kwargs)
|
|
||||||
response.raise_for_status()
|
|
||||||
return response.json()
|
|
||||||
except httpx.HTTPStatusError:
|
|
||||||
self._handle_error(response)
|
|
||||||
except httpx.RequestError as e:
|
|
||||||
raise GiteaClientError(
|
|
||||||
f"Request failed: {e}"
|
|
||||||
) from e
|
|
||||||
|
|
||||||
async def post(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
|
|
||||||
"""Make POST request to Gitea API.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
path: API endpoint path.
|
|
||||||
json: JSON data to send in request body.
|
|
||||||
**kwargs: Additional arguments for httpx request.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
dict: JSON response data.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
GiteaClientError: If request fails.
|
|
||||||
"""
|
|
||||||
if not self._client:
|
|
||||||
raise GiteaClientError(
|
|
||||||
"Client not initialized. Use 'async with' context manager."
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
response = await self._client.post(path, json=json, **kwargs)
|
|
||||||
response.raise_for_status()
|
|
||||||
return response.json()
|
|
||||||
except httpx.HTTPStatusError:
|
|
||||||
self._handle_error(response)
|
|
||||||
except httpx.RequestError as e:
|
|
||||||
raise GiteaClientError(
|
|
||||||
f"Request failed: {e}"
|
|
||||||
) from e
|
|
||||||
|
|
||||||
async def patch(self, path: str, json: Optional[dict[str, Any]] = None, **kwargs) -> dict[str, Any]:
|
|
||||||
"""Make PATCH request to Gitea API.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
path: API endpoint path.
|
|
||||||
json: JSON data to send in request body.
|
|
||||||
**kwargs: Additional arguments for httpx request.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
dict: JSON response data.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
GiteaClientError: If request fails.
|
|
||||||
"""
|
|
||||||
if not self._client:
|
|
||||||
raise GiteaClientError(
|
|
||||||
"Client not initialized. Use 'async with' context manager."
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
response = await self._client.patch(path, json=json, **kwargs)
|
|
||||||
response.raise_for_status()
|
|
||||||
return response.json()
|
|
||||||
except httpx.HTTPStatusError:
|
|
||||||
self._handle_error(response)
|
|
||||||
except httpx.RequestError as e:
|
|
||||||
raise GiteaClientError(
|
|
||||||
f"Request failed: {e}"
|
|
||||||
) from e
|
|
||||||
|
|
||||||
async def delete(self, path: str, **kwargs) -> Optional[dict[str, Any]]:
|
|
||||||
"""Make DELETE request to Gitea API.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
path: API endpoint path.
|
|
||||||
**kwargs: Additional arguments for httpx request.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
dict or None: JSON response data if available, None for 204 responses.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
GiteaClientError: If request fails.
|
|
||||||
"""
|
|
||||||
if not self._client:
|
|
||||||
raise GiteaClientError(
|
|
||||||
"Client not initialized. Use 'async with' context manager."
|
|
||||||
)
|
|
||||||
|
|
||||||
try:
|
|
||||||
response = await self._client.delete(path, **kwargs)
|
|
||||||
response.raise_for_status()
|
|
||||||
|
|
||||||
# DELETE requests may return 204 No Content
|
|
||||||
if response.status_code == 204:
|
|
||||||
return None
|
|
||||||
|
|
||||||
return response.json()
|
|
||||||
except httpx.HTTPStatusError:
|
|
||||||
self._handle_error(response)
|
|
||||||
except httpx.RequestError as e:
|
|
||||||
raise GiteaClientError(
|
|
||||||
f"Request failed: {e}"
|
|
||||||
) from e
|
|
||||||
@@ -1,173 +0,0 @@
|
|||||||
"""MCP server implementation for Gitea API integration."""
|
|
||||||
|
|
||||||
import asyncio
|
|
||||||
import argparse
|
|
||||||
from mcp.server import Server
|
|
||||||
from mcp.server.stdio import stdio_server
|
|
||||||
from mcp.types import Tool, TextContent
|
|
||||||
|
|
||||||
from . import __version__
|
|
||||||
from .auth import AuthConfig
|
|
||||||
from .client import GiteaClient, GiteaClientError
|
|
||||||
from .tools import (
|
|
||||||
get_issue_tools,
|
|
||||||
handle_issue_tool,
|
|
||||||
get_label_tools,
|
|
||||||
handle_label_tool,
|
|
||||||
get_milestone_tools,
|
|
||||||
handle_milestone_tool,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
# Global client instance
|
|
||||||
gitea_client: GiteaClient | None = None
|
|
||||||
|
|
||||||
|
|
||||||
async def serve() -> None:
|
|
||||||
"""Run the MCP server."""
|
|
||||||
server = Server("gitea-mcp")
|
|
||||||
|
|
||||||
# Initialize authentication config
|
|
||||||
try:
|
|
||||||
config = AuthConfig()
|
|
||||||
except ValueError as e:
|
|
||||||
print(f"Configuration error: {e}")
|
|
||||||
raise
|
|
||||||
|
|
||||||
# Initialize Gitea client
|
|
||||||
global gitea_client
|
|
||||||
gitea_client = GiteaClient(config)
|
|
||||||
|
|
||||||
@server.list_tools()
|
|
||||||
async def list_tools() -> list[Tool]:
|
|
||||||
"""List available MCP tools.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Available tools including issue, label, and milestone operations.
|
|
||||||
"""
|
|
||||||
# Get issue, label, and milestone tools
|
|
||||||
tools = get_issue_tools()
|
|
||||||
tools.extend(get_label_tools())
|
|
||||||
tools.extend(get_milestone_tools())
|
|
||||||
|
|
||||||
# Placeholder for future tools (PR tools, etc.)
|
|
||||||
tools.extend([
|
|
||||||
Tool(
|
|
||||||
name="list_repositories",
|
|
||||||
description="List repositories in an organization (coming soon)",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"org": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Organization name",
|
|
||||||
}
|
|
||||||
},
|
|
||||||
"required": ["org"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="create_pull_request",
|
|
||||||
description="Create a new pull request (coming soon)",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"title": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Pull request title",
|
|
||||||
},
|
|
||||||
"head": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Source branch",
|
|
||||||
},
|
|
||||||
"base": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Target branch",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "title", "head", "base"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
])
|
|
||||||
|
|
||||||
return tools
|
|
||||||
|
|
||||||
@server.call_tool()
|
|
||||||
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
|
||||||
"""Handle tool calls.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
name: Tool name.
|
|
||||||
arguments: Tool arguments.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Tool response.
|
|
||||||
"""
|
|
||||||
# Handle issue tools
|
|
||||||
if name.startswith("gitea_") and any(
|
|
||||||
name.endswith(suffix) for suffix in ["_issues", "_issue"]
|
|
||||||
):
|
|
||||||
return await handle_issue_tool(name, arguments, gitea_client)
|
|
||||||
|
|
||||||
# Handle label tools
|
|
||||||
if name.startswith("gitea_") and any(
|
|
||||||
name.endswith(suffix) for suffix in ["_labels", "_label"]
|
|
||||||
):
|
|
||||||
return await handle_label_tool(gitea_client, name, arguments)
|
|
||||||
|
|
||||||
# Handle milestone tools
|
|
||||||
if name.startswith("gitea_") and any(
|
|
||||||
name.endswith(suffix) for suffix in ["_milestones", "_milestone"]
|
|
||||||
):
|
|
||||||
return await handle_milestone_tool(name, arguments, gitea_client)
|
|
||||||
|
|
||||||
# Placeholder for other tools
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Tool '{name}' is not yet implemented.",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
# Run the server using stdio transport
|
|
||||||
async with stdio_server() as (read_stream, write_stream):
|
|
||||||
await server.run(
|
|
||||||
read_stream,
|
|
||||||
write_stream,
|
|
||||||
server.create_initialization_options(),
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
def main() -> None:
|
|
||||||
"""Main entry point with CLI argument parsing."""
|
|
||||||
parser = argparse.ArgumentParser(
|
|
||||||
description="Gitea MCP Server - MCP server for Gitea API integration"
|
|
||||||
)
|
|
||||||
parser.add_argument(
|
|
||||||
"--version",
|
|
||||||
action="version",
|
|
||||||
version=f"gitea-mcp {__version__}",
|
|
||||||
)
|
|
||||||
|
|
||||||
args = parser.parse_args()
|
|
||||||
|
|
||||||
# Run the server
|
|
||||||
try:
|
|
||||||
asyncio.run(serve())
|
|
||||||
except KeyboardInterrupt:
|
|
||||||
print("\nServer stopped by user")
|
|
||||||
except Exception as e:
|
|
||||||
print(f"Server error: {e}")
|
|
||||||
raise
|
|
||||||
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
main()
|
|
||||||
@@ -1,14 +0,0 @@
|
|||||||
"""Gitea MCP tools package."""
|
|
||||||
|
|
||||||
from .issues import get_issue_tools, handle_issue_tool
|
|
||||||
from .labels import get_label_tools, handle_label_tool
|
|
||||||
from .milestones import get_milestone_tools, handle_milestone_tool
|
|
||||||
|
|
||||||
__all__ = [
|
|
||||||
"get_issue_tools",
|
|
||||||
"handle_issue_tool",
|
|
||||||
"get_label_tools",
|
|
||||||
"handle_label_tool",
|
|
||||||
"get_milestone_tools",
|
|
||||||
"handle_milestone_tool",
|
|
||||||
]
|
|
||||||
@@ -1,392 +0,0 @@
|
|||||||
"""Gitea issue operations tools for MCP server."""
|
|
||||||
|
|
||||||
from typing import Any, Optional
|
|
||||||
from mcp.types import Tool, TextContent
|
|
||||||
|
|
||||||
from ..client import GiteaClient, GiteaClientError
|
|
||||||
|
|
||||||
|
|
||||||
def get_issue_tools() -> list[Tool]:
|
|
||||||
"""Get list of issue operation tools.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[Tool]: List of MCP tools for issue operations.
|
|
||||||
"""
|
|
||||||
return [
|
|
||||||
Tool(
|
|
||||||
name="gitea_list_issues",
|
|
||||||
description="List issues in a Gitea repository with optional filters",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"state": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Filter by state: open, closed, or all",
|
|
||||||
"enum": ["open", "closed", "all"],
|
|
||||||
"default": "open",
|
|
||||||
},
|
|
||||||
"labels": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Comma-separated list of label names to filter by",
|
|
||||||
},
|
|
||||||
"milestone": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Milestone name to filter by",
|
|
||||||
},
|
|
||||||
"page": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Page number for pagination (default: 1)",
|
|
||||||
"default": 1,
|
|
||||||
},
|
|
||||||
"limit": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Number of issues per page (default: 30)",
|
|
||||||
"default": 30,
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="gitea_get_issue",
|
|
||||||
description="Get details of a specific issue by number",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"index": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Issue number/index",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "index"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="gitea_create_issue",
|
|
||||||
description="Create a new issue in a Gitea repository",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"title": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Issue title",
|
|
||||||
},
|
|
||||||
"body": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Issue body/description",
|
|
||||||
},
|
|
||||||
"labels": {
|
|
||||||
"type": "array",
|
|
||||||
"items": {"type": "integer"},
|
|
||||||
"description": "Array of label IDs to assign",
|
|
||||||
},
|
|
||||||
"milestone": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Milestone ID to assign",
|
|
||||||
},
|
|
||||||
"assignees": {
|
|
||||||
"type": "array",
|
|
||||||
"items": {"type": "string"},
|
|
||||||
"description": "Array of usernames to assign",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "title"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="gitea_update_issue",
|
|
||||||
description="Update an existing issue in a Gitea repository",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"index": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Issue number/index",
|
|
||||||
},
|
|
||||||
"title": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "New issue title",
|
|
||||||
},
|
|
||||||
"body": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "New issue body/description",
|
|
||||||
},
|
|
||||||
"state": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Issue state: open or closed",
|
|
||||||
"enum": ["open", "closed"],
|
|
||||||
},
|
|
||||||
"labels": {
|
|
||||||
"type": "array",
|
|
||||||
"items": {"type": "integer"},
|
|
||||||
"description": "Array of label IDs to assign (replaces existing)",
|
|
||||||
},
|
|
||||||
"milestone": {
|
|
||||||
"type": "integer",
|
|
||||||
"description": "Milestone ID to assign",
|
|
||||||
},
|
|
||||||
"assignees": {
|
|
||||||
"type": "array",
|
|
||||||
"items": {"type": "string"},
|
|
||||||
"description": "Array of usernames to assign (replaces existing)",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "index"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def handle_issue_tool(
|
|
||||||
name: str, arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Handle issue tool calls.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
name: Tool name.
|
|
||||||
arguments: Tool arguments.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Tool response.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
if name == "gitea_list_issues":
|
|
||||||
return await _list_issues(arguments, client)
|
|
||||||
elif name == "gitea_get_issue":
|
|
||||||
return await _get_issue(arguments, client)
|
|
||||||
elif name == "gitea_create_issue":
|
|
||||||
return await _create_issue(arguments, client)
|
|
||||||
elif name == "gitea_update_issue":
|
|
||||||
return await _update_issue(arguments, client)
|
|
||||||
else:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Unknown issue tool: {name}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
except GiteaClientError as e:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Error: {str(e)}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def _list_issues(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""List issues in a repository.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, and optional filters.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: List of issues.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
state = arguments.get("state", "open")
|
|
||||||
labels = arguments.get("labels")
|
|
||||||
milestone = arguments.get("milestone")
|
|
||||||
page = arguments.get("page", 1)
|
|
||||||
limit = arguments.get("limit", 30)
|
|
||||||
|
|
||||||
params = {
|
|
||||||
"state": state,
|
|
||||||
"page": page,
|
|
||||||
"limit": limit,
|
|
||||||
}
|
|
||||||
|
|
||||||
if labels:
|
|
||||||
params["labels"] = labels
|
|
||||||
|
|
||||||
if milestone:
|
|
||||||
params["milestone"] = milestone
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
issues = await client.get(f"/repos/{owner}/{repo}/issues", params=params)
|
|
||||||
|
|
||||||
# Format response
|
|
||||||
if not issues:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"No {state} issues found in {owner}/{repo}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
result = f"Found {len(issues)} {state} issue(s) in {owner}/{repo}:\n\n"
|
|
||||||
for issue in issues:
|
|
||||||
result += f"#{issue['number']} - {issue['title']}\n"
|
|
||||||
result += f" State: {issue['state']}\n"
|
|
||||||
if issue.get('labels'):
|
|
||||||
labels_str = ", ".join([label['name'] for label in issue['labels']])
|
|
||||||
result += f" Labels: {labels_str}\n"
|
|
||||||
if issue.get('milestone'):
|
|
||||||
result += f" Milestone: {issue['milestone']['title']}\n"
|
|
||||||
result += f" Created: {issue['created_at']}\n"
|
|
||||||
result += f" Updated: {issue['updated_at']}\n\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
|
|
||||||
|
|
||||||
async def _get_issue(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Get a specific issue by number.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, and index.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Issue details.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
index = arguments["index"]
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
issue = await client.get(f"/repos/{owner}/{repo}/issues/{index}")
|
|
||||||
|
|
||||||
# Format response
|
|
||||||
result = f"Issue #{issue['number']}: {issue['title']}\n\n"
|
|
||||||
result += f"State: {issue['state']}\n"
|
|
||||||
result += f"Created: {issue['created_at']}\n"
|
|
||||||
result += f"Updated: {issue['updated_at']}\n"
|
|
||||||
|
|
||||||
if issue.get('labels'):
|
|
||||||
labels_str = ", ".join([label['name'] for label in issue['labels']])
|
|
||||||
result += f"Labels: {labels_str}\n"
|
|
||||||
|
|
||||||
if issue.get('milestone'):
|
|
||||||
result += f"Milestone: {issue['milestone']['title']}\n"
|
|
||||||
|
|
||||||
if issue.get('assignees'):
|
|
||||||
assignees_str = ", ".join([user['login'] for user in issue['assignees']])
|
|
||||||
result += f"Assignees: {assignees_str}\n"
|
|
||||||
|
|
||||||
result += f"\nBody:\n{issue.get('body', '(no description)')}\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
|
|
||||||
|
|
||||||
async def _create_issue(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Create a new issue.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, title, and optional fields.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Created issue details.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
|
|
||||||
data = {
|
|
||||||
"title": arguments["title"],
|
|
||||||
}
|
|
||||||
|
|
||||||
if "body" in arguments:
|
|
||||||
data["body"] = arguments["body"]
|
|
||||||
|
|
||||||
if "labels" in arguments:
|
|
||||||
data["labels"] = arguments["labels"]
|
|
||||||
|
|
||||||
if "milestone" in arguments:
|
|
||||||
data["milestone"] = arguments["milestone"]
|
|
||||||
|
|
||||||
if "assignees" in arguments:
|
|
||||||
data["assignees"] = arguments["assignees"]
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
issue = await client.post(f"/repos/{owner}/{repo}/issues", json=data)
|
|
||||||
|
|
||||||
result = f"Created issue #{issue['number']}: {issue['title']}\n"
|
|
||||||
result += f"URL: {issue.get('html_url', 'N/A')}\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
|
|
||||||
|
|
||||||
async def _update_issue(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Update an existing issue.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, index, and fields to update.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Updated issue details.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
index = arguments["index"]
|
|
||||||
|
|
||||||
data = {}
|
|
||||||
|
|
||||||
if "title" in arguments:
|
|
||||||
data["title"] = arguments["title"]
|
|
||||||
|
|
||||||
if "body" in arguments:
|
|
||||||
data["body"] = arguments["body"]
|
|
||||||
|
|
||||||
if "state" in arguments:
|
|
||||||
data["state"] = arguments["state"]
|
|
||||||
|
|
||||||
if "labels" in arguments:
|
|
||||||
data["labels"] = arguments["labels"]
|
|
||||||
|
|
||||||
if "milestone" in arguments:
|
|
||||||
data["milestone"] = arguments["milestone"]
|
|
||||||
|
|
||||||
if "assignees" in arguments:
|
|
||||||
data["assignees"] = arguments["assignees"]
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
issue = await client.patch(f"/repos/{owner}/{repo}/issues/{index}", json=data)
|
|
||||||
|
|
||||||
result = f"Updated issue #{issue['number']}: {issue['title']}\n"
|
|
||||||
result += f"State: {issue['state']}\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
@@ -1,170 +0,0 @@
|
|||||||
"""Gitea label operations MCP tools."""
|
|
||||||
|
|
||||||
from mcp.types import Tool, TextContent
|
|
||||||
|
|
||||||
from ..client import GiteaClient, GiteaClientError
|
|
||||||
|
|
||||||
|
|
||||||
def get_label_tools() -> list[Tool]:
|
|
||||||
"""Get label operation tool definitions.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Tool definitions for label operations.
|
|
||||||
"""
|
|
||||||
return [
|
|
||||||
Tool(
|
|
||||||
name="gitea_list_labels",
|
|
||||||
description="List all labels in a Gitea repository",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (user or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="gitea_create_label",
|
|
||||||
description="Create a new label in a Gitea repository",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (user or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"name": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Label name",
|
|
||||||
},
|
|
||||||
"color": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Label color (hex without #, e.g., 'ff0000' for red)",
|
|
||||||
},
|
|
||||||
"description": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Label description (optional)",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "name", "color"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def handle_label_tool(
|
|
||||||
client: GiteaClient, name: str, arguments: dict
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Handle label tool execution.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client: Gitea client instance.
|
|
||||||
name: Tool name.
|
|
||||||
arguments: Tool arguments.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Tool response content.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
async with client:
|
|
||||||
if name == "gitea_list_labels":
|
|
||||||
return await _list_labels(client, arguments)
|
|
||||||
elif name == "gitea_create_label":
|
|
||||||
return await _create_label(client, arguments)
|
|
||||||
else:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Unknown label tool: {name}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
except GiteaClientError as e:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Gitea API error: {e}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def _list_labels(client: GiteaClient, arguments: dict) -> list[TextContent]:
|
|
||||||
"""List labels in a repository.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client: Gitea client instance.
|
|
||||||
arguments: Tool arguments with owner and repo.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Label listing response.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
|
|
||||||
labels = await client.get(f"/repos/{owner}/{repo}/labels")
|
|
||||||
|
|
||||||
if not labels:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"No labels found in {owner}/{repo}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
# Format labels for display
|
|
||||||
lines = [f"Labels in {owner}/{repo}:", ""]
|
|
||||||
for label in labels:
|
|
||||||
color = label.get("color", "")
|
|
||||||
desc = label.get("description", "")
|
|
||||||
desc_text = f" - {desc}" if desc else ""
|
|
||||||
lines.append(f" • {label['name']} (#{color}){desc_text}")
|
|
||||||
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text="\n".join(lines),
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def _create_label(client: GiteaClient, arguments: dict) -> list[TextContent]:
|
|
||||||
"""Create a new label.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
client: Gitea client instance.
|
|
||||||
arguments: Tool arguments with owner, repo, name, color, and optional description.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list: Creation response.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
name = arguments["name"]
|
|
||||||
color = arguments["color"].lstrip("#") # Remove # if present
|
|
||||||
description = arguments.get("description", "")
|
|
||||||
|
|
||||||
payload = {
|
|
||||||
"name": name,
|
|
||||||
"color": color,
|
|
||||||
}
|
|
||||||
if description:
|
|
||||||
payload["description"] = description
|
|
||||||
|
|
||||||
label = await client.post(f"/repos/{owner}/{repo}/labels", payload)
|
|
||||||
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Created label '{label['name']}' (#{label['color']}) in {owner}/{repo}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
@@ -1,190 +0,0 @@
|
|||||||
"""Gitea milestone operations tools for MCP server."""
|
|
||||||
|
|
||||||
from typing import Any, Optional
|
|
||||||
from mcp.types import Tool, TextContent
|
|
||||||
|
|
||||||
from ..client import GiteaClient, GiteaClientError
|
|
||||||
|
|
||||||
|
|
||||||
def get_milestone_tools() -> list[Tool]:
|
|
||||||
"""Get list of milestone operation tools.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[Tool]: List of MCP tools for milestone operations.
|
|
||||||
"""
|
|
||||||
return [
|
|
||||||
Tool(
|
|
||||||
name="gitea_list_milestones",
|
|
||||||
description="List milestones in a Gitea repository with optional state filter",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"state": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Filter by state: open, closed, or all (default: open)",
|
|
||||||
"enum": ["open", "closed", "all"],
|
|
||||||
"default": "open",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
Tool(
|
|
||||||
name="gitea_create_milestone",
|
|
||||||
description="Create a new milestone in a Gitea repository",
|
|
||||||
inputSchema={
|
|
||||||
"type": "object",
|
|
||||||
"properties": {
|
|
||||||
"owner": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository owner (username or organization)",
|
|
||||||
},
|
|
||||||
"repo": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Repository name",
|
|
||||||
},
|
|
||||||
"title": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Milestone title",
|
|
||||||
},
|
|
||||||
"description": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Milestone description (optional)",
|
|
||||||
},
|
|
||||||
"due_on": {
|
|
||||||
"type": "string",
|
|
||||||
"description": "Due date in ISO 8601 format, e.g., '2024-12-31T23:59:59Z' (optional)",
|
|
||||||
},
|
|
||||||
},
|
|
||||||
"required": ["owner", "repo", "title"],
|
|
||||||
},
|
|
||||||
),
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def handle_milestone_tool(
|
|
||||||
name: str, arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Handle milestone tool calls.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
name: Tool name.
|
|
||||||
arguments: Tool arguments.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Tool response.
|
|
||||||
"""
|
|
||||||
try:
|
|
||||||
if name == "gitea_list_milestones":
|
|
||||||
return await _list_milestones(arguments, client)
|
|
||||||
elif name == "gitea_create_milestone":
|
|
||||||
return await _create_milestone(arguments, client)
|
|
||||||
else:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Unknown milestone tool: {name}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
except GiteaClientError as e:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"Error: {str(e)}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def _list_milestones(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""List milestones in a repository.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, and optional state filter.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: List of milestones.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
state = arguments.get("state", "open")
|
|
||||||
|
|
||||||
params = {"state": state}
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
milestones = await client.get(
|
|
||||||
f"/repos/{owner}/{repo}/milestones", params=params
|
|
||||||
)
|
|
||||||
|
|
||||||
# Format response
|
|
||||||
if not milestones:
|
|
||||||
return [
|
|
||||||
TextContent(
|
|
||||||
type="text",
|
|
||||||
text=f"No {state} milestones found in {owner}/{repo}",
|
|
||||||
)
|
|
||||||
]
|
|
||||||
|
|
||||||
result = f"Found {len(milestones)} {state} milestone(s) in {owner}/{repo}:\n\n"
|
|
||||||
for milestone in milestones:
|
|
||||||
result += f"{milestone.get('title', 'Untitled')}\n"
|
|
||||||
result += f" State: {milestone.get('state', 'unknown')}\n"
|
|
||||||
if milestone.get("description"):
|
|
||||||
result += f" Description: {milestone['description']}\n"
|
|
||||||
if milestone.get("due_on"):
|
|
||||||
result += f" Due: {milestone['due_on']}\n"
|
|
||||||
result += f" Open Issues: {milestone.get('open_issues', 0)}\n"
|
|
||||||
result += f" Closed Issues: {milestone.get('closed_issues', 0)}\n"
|
|
||||||
result += f" Created: {milestone.get('created_at', 'N/A')}\n\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
|
|
||||||
|
|
||||||
async def _create_milestone(
|
|
||||||
arguments: dict[str, Any], client: GiteaClient
|
|
||||||
) -> list[TextContent]:
|
|
||||||
"""Create a new milestone.
|
|
||||||
|
|
||||||
Args:
|
|
||||||
arguments: Tool arguments containing owner, repo, title, and optional fields.
|
|
||||||
client: Gitea API client instance.
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
list[TextContent]: Created milestone details.
|
|
||||||
"""
|
|
||||||
owner = arguments["owner"]
|
|
||||||
repo = arguments["repo"]
|
|
||||||
|
|
||||||
data = {
|
|
||||||
"title": arguments["title"],
|
|
||||||
}
|
|
||||||
|
|
||||||
if "description" in arguments:
|
|
||||||
data["description"] = arguments["description"]
|
|
||||||
|
|
||||||
if "due_on" in arguments:
|
|
||||||
data["due_on"] = arguments["due_on"]
|
|
||||||
|
|
||||||
async with client:
|
|
||||||
milestone = await client.post(
|
|
||||||
f"/repos/{owner}/{repo}/milestones", json=data
|
|
||||||
)
|
|
||||||
|
|
||||||
result = f"Created milestone: {milestone['title']}\n"
|
|
||||||
if milestone.get("description"):
|
|
||||||
result += f"Description: {milestone['description']}\n"
|
|
||||||
if milestone.get("due_on"):
|
|
||||||
result += f"Due: {milestone['due_on']}\n"
|
|
||||||
|
|
||||||
return [TextContent(type="text", text=result)]
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
"""Tests for Gitea MCP server."""
|
|
||||||
@@ -1,56 +0,0 @@
|
|||||||
"""Tests for authentication module."""
|
|
||||||
|
|
||||||
import os
|
|
||||||
import pytest
|
|
||||||
from gitea_mcp.auth import AuthConfig
|
|
||||||
|
|
||||||
|
|
||||||
def test_auth_config_success(monkeypatch):
|
|
||||||
"""Test successful authentication configuration."""
|
|
||||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
|
||||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
|
||||||
|
|
||||||
config = AuthConfig()
|
|
||||||
|
|
||||||
assert config.api_url == "http://gitea.example.com/api/v1"
|
|
||||||
assert config.api_token == "test_token_123"
|
|
||||||
|
|
||||||
|
|
||||||
def test_auth_config_removes_trailing_slash(monkeypatch):
|
|
||||||
"""Test that trailing slash is removed from URL."""
|
|
||||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1/")
|
|
||||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
|
||||||
|
|
||||||
config = AuthConfig()
|
|
||||||
|
|
||||||
assert config.api_url == "http://gitea.example.com/api/v1"
|
|
||||||
|
|
||||||
|
|
||||||
def test_auth_config_missing_url(monkeypatch):
|
|
||||||
"""Test error when GITEA_API_URL is missing."""
|
|
||||||
monkeypatch.delenv("GITEA_API_URL", raising=False)
|
|
||||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="GITEA_API_URL"):
|
|
||||||
AuthConfig()
|
|
||||||
|
|
||||||
|
|
||||||
def test_auth_config_missing_token(monkeypatch):
|
|
||||||
"""Test error when GITEA_API_TOKEN is missing."""
|
|
||||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
|
||||||
monkeypatch.delenv("GITEA_API_TOKEN", raising=False)
|
|
||||||
|
|
||||||
with pytest.raises(ValueError, match="GITEA_API_TOKEN"):
|
|
||||||
AuthConfig()
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_auth_headers(monkeypatch):
|
|
||||||
"""Test authentication headers generation."""
|
|
||||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
|
||||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
|
||||||
|
|
||||||
config = AuthConfig()
|
|
||||||
headers = config.get_auth_headers()
|
|
||||||
|
|
||||||
assert headers["Authorization"] == "token test_token_123"
|
|
||||||
assert headers["Content-Type"] == "application/json"
|
|
||||||
@@ -1,93 +0,0 @@
|
|||||||
"""Tests for Gitea API client."""
|
|
||||||
|
|
||||||
import pytest
|
|
||||||
import httpx
|
|
||||||
from unittest.mock import AsyncMock, MagicMock
|
|
||||||
from gitea_mcp.auth import AuthConfig
|
|
||||||
from gitea_mcp.client import (
|
|
||||||
GiteaClient,
|
|
||||||
GiteaClientError,
|
|
||||||
GiteaAuthError,
|
|
||||||
GiteaNotFoundError,
|
|
||||||
GiteaServerError,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def mock_config(monkeypatch):
|
|
||||||
"""Create mock authentication config."""
|
|
||||||
monkeypatch.setenv("GITEA_API_URL", "http://gitea.example.com/api/v1")
|
|
||||||
monkeypatch.setenv("GITEA_API_TOKEN", "test_token_123")
|
|
||||||
return AuthConfig()
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_client_initialization(mock_config):
|
|
||||||
"""Test client initialization."""
|
|
||||||
client = GiteaClient(mock_config, timeout=10.0)
|
|
||||||
assert client.config == mock_config
|
|
||||||
assert client.timeout == 10.0
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_client_context_manager(mock_config):
|
|
||||||
"""Test client as async context manager."""
|
|
||||||
async with GiteaClient(mock_config) as client:
|
|
||||||
assert client._client is not None
|
|
||||||
assert isinstance(client._client, httpx.AsyncClient)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_client_without_context_manager_raises_error(mock_config):
|
|
||||||
"""Test that using client without context manager raises error."""
|
|
||||||
client = GiteaClient(mock_config)
|
|
||||||
|
|
||||||
with pytest.raises(GiteaClientError, match="not initialized"):
|
|
||||||
await client.get("/test")
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_handle_error_401(mock_config):
|
|
||||||
"""Test handling 401 authentication error."""
|
|
||||||
response = MagicMock()
|
|
||||||
response.status_code = 401
|
|
||||||
|
|
||||||
async with GiteaClient(mock_config) as client:
|
|
||||||
with pytest.raises(GiteaAuthError, match="Authentication failed"):
|
|
||||||
client._handle_error(response)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_handle_error_403(mock_config):
|
|
||||||
"""Test handling 403 forbidden error."""
|
|
||||||
response = MagicMock()
|
|
||||||
response.status_code = 403
|
|
||||||
|
|
||||||
async with GiteaClient(mock_config) as client:
|
|
||||||
with pytest.raises(GiteaAuthError, match="Access forbidden"):
|
|
||||||
client._handle_error(response)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_handle_error_404(mock_config):
|
|
||||||
"""Test handling 404 not found error."""
|
|
||||||
response = MagicMock()
|
|
||||||
response.status_code = 404
|
|
||||||
response.request = MagicMock()
|
|
||||||
response.request.url = "http://gitea.example.com/api/v1/test"
|
|
||||||
|
|
||||||
async with GiteaClient(mock_config) as client:
|
|
||||||
with pytest.raises(GiteaNotFoundError, match="not found"):
|
|
||||||
client._handle_error(response)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.asyncio
|
|
||||||
async def test_handle_error_500(mock_config):
|
|
||||||
"""Test handling 500 server error."""
|
|
||||||
response = MagicMock()
|
|
||||||
response.status_code = 500
|
|
||||||
response.text = "Internal Server Error"
|
|
||||||
|
|
||||||
async with GiteaClient(mock_config) as client:
|
|
||||||
with pytest.raises(GiteaServerError, match="server error"):
|
|
||||||
client._handle_error(response)
|
|
||||||
Reference in New Issue
Block a user