420 lines
17 KiB
Python
420 lines
17 KiB
Python
import os
|
|
import logging
|
|
from typing import Any, Optional
|
|
import httpx
|
|
from fastapi import FastAPI, Request, HTTPException, Header
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from fastapi.responses import JSONResponse
|
|
from dotenv import load_dotenv
|
|
|
|
from mcp.server import Server
|
|
from mcp.server.sse import SseServerTransport
|
|
from mcp.types import Tool, TextContent
|
|
|
|
# Load environment variables
|
|
load_dotenv()
|
|
|
|
# Logging configuration
|
|
logging.basicConfig(level=logging.INFO)
|
|
logger = logging.getLogger("gitea-mcp-adapter")
|
|
|
|
GITEA_URL = os.getenv("GITEA_URL", "https://git.carmona.digital").rstrip("/")
|
|
GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
|
|
|
|
app = FastAPI(
|
|
title="Gitea MCP Adapter",
|
|
description="Proxy wrapping Gitea REST API as an MCP server",
|
|
version="1.0.0"
|
|
)
|
|
|
|
# Enable CORS
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=["*"],
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Initialize MCP Server
|
|
mcp_server = Server("gitea-mcp-adapter")
|
|
|
|
# Helper to get HTTPlient with Gitea headers
|
|
def get_gitea_client(auth_header: Optional[str] = None) -> httpx.AsyncClient:
|
|
headers = {
|
|
"Accept": "application/json",
|
|
"Content-Type": "application/json"
|
|
}
|
|
|
|
# Use token from Authorization header if present, otherwise fallback to GITEA_TOKEN
|
|
token = GITEA_TOKEN
|
|
if auth_header and auth_header.startswith("Bearer "):
|
|
token = auth_header.split(" ")[1]
|
|
|
|
if token:
|
|
headers["Authorization"] = f"token {token}"
|
|
|
|
return httpx.AsyncClient(base_url=GITEA_URL, headers=headers, timeout=15.0)
|
|
|
|
# Define Gitea MCP Tools
|
|
@mcp_server.list_tools()
|
|
async def list_tools() -> list[Tool]:
|
|
return [
|
|
Tool(
|
|
name="issues_abiertos",
|
|
description="List open issues in a Gitea repository. Supports filtering by labels.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"labels": {"type": "string", "description": "Comma-separated list of labels to filter (optional)"}
|
|
},
|
|
"required": ["repo"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="commits_recientes",
|
|
description="Get the last N commits of a branch in a Gitea repository.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"limit": {"type": "integer", "description": "Number of commits to return (default: 10)", "default": 10},
|
|
"branch": {"type": "string", "description": "Branch name (default: main)", "default": "main"}
|
|
},
|
|
"required": ["repo"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="estado_actions",
|
|
description="Get the last workflow runs of Gitea Actions for a Gitea repository.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"limit": {"type": "integer", "description": "Number of runs to return (default: 5)", "default": 5}
|
|
},
|
|
"required": ["repo"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="crear_issue",
|
|
description="Create a new issue in a Gitea repository.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"title": {"type": "string", "description": "Issue title"},
|
|
"body": {"type": "string", "description": "Issue description/body (optional)"},
|
|
"labels": {
|
|
"type": "array",
|
|
"items": {"type": "string"},
|
|
"description": "List of label names to apply (optional)"
|
|
}
|
|
},
|
|
"required": ["repo", "title"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="comentar_issue",
|
|
description="Add a comment to an existing Gitea issue.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"issue_number": {"type": "integer", "description": "The issue number"},
|
|
"body": {"type": "string", "description": "The comment body text"}
|
|
},
|
|
"required": ["repo", "issue_number", "body"]
|
|
}
|
|
),
|
|
Tool(
|
|
name="crear_pull_request",
|
|
description="Create a new pull request in a Gitea repository.",
|
|
inputSchema={
|
|
"type": "object",
|
|
"properties": {
|
|
"repo": {"type": "string", "description": "Repository name"},
|
|
"owner": {"type": "string", "description": "Repository owner/organization (default: estamos)", "default": "estamos"},
|
|
"title": {"type": "string", "description": "PR title"},
|
|
"body": {"type": "string", "description": "PR description/body (optional)"},
|
|
"head": {"type": "string", "description": "The branch where changes are implemented (source branch)"},
|
|
"base": {"type": "string", "description": "The branch into which changes should be merged (default: main)", "default": "main"}
|
|
},
|
|
"required": ["repo", "title", "head"]
|
|
}
|
|
)
|
|
]
|
|
|
|
# Execute Gitea MCP Tools
|
|
@mcp_server.call_tool()
|
|
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
|
|
# Note: Since the call_tool is run within the MCP context, we can extract the Authorization header
|
|
# if we pass it through context, but since MCP call_tool doesn't directly receive HTTP headers,
|
|
# we default to the GITEA_TOKEN environment variable.
|
|
async with get_gitea_client() as client:
|
|
try:
|
|
owner = arguments.get("owner", "estamos")
|
|
repo = arguments.get("repo")
|
|
|
|
if name == "issues_abiertos":
|
|
params = {"state": "open", "type": "issues", "limit": 50}
|
|
if arguments.get("labels"):
|
|
params["labels"] = arguments["labels"]
|
|
|
|
response = await client.get(f"/api/v1/repos/{owner}/{repo}/issues", params=params)
|
|
response.raise_for_status()
|
|
issues = response.json()
|
|
|
|
rows = [
|
|
{
|
|
"number": i.get("number"),
|
|
"title": i.get("title"),
|
|
"state": i.get("state"),
|
|
"labels": [lb.get("name") for lb in (i.get("labels") or [])],
|
|
"assignee": (i.get("assignee") or {}).get("login"),
|
|
"created_at": i.get("created_at"),
|
|
"html_url": i.get("html_url")
|
|
}
|
|
for i in (issues if isinstance(issues, list) else [])
|
|
]
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({"rows": rows, "count": len(rows), "repo": repo}))]
|
|
|
|
elif name == "commits_recientes":
|
|
limit = min(int(arguments.get("limit", 10)), 50)
|
|
branch = arguments.get("branch", "main")
|
|
|
|
params = {"sha": branch, "limit": limit}
|
|
response = await client.get(f"/api/v1/repos/{owner}/{repo}/commits", params=params)
|
|
response.raise_for_status()
|
|
commits = response.json()
|
|
|
|
rows = [
|
|
{
|
|
"sha": c.get("sha", "")[:8],
|
|
"message": (c.get("commit") or {}).get("message", "").split("\n")[0],
|
|
"author": ((c.get("commit") or {}).get("author") or {}).get("name"),
|
|
"date": ((c.get("commit") or {}).get("author") or {}).get("date")
|
|
}
|
|
for c in (commits if isinstance(commits, list) else [])
|
|
]
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({"rows": rows, "count": len(rows), "repo": repo, "branch": branch}))]
|
|
|
|
elif name == "estado_actions":
|
|
limit = min(int(arguments.get("limit", 5)), 20)
|
|
|
|
try:
|
|
response = await client.get(f"/api/v1/repos/{owner}/{repo}/actions/runs", params={"limit": limit})
|
|
response.raise_for_status()
|
|
data = response.json()
|
|
except httpx.HTTPStatusError as exc:
|
|
if exc.response.status_code in (403, 404):
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({"available": False, "repo": repo}))]
|
|
raise
|
|
|
|
runs = data.get("workflow_runs") or [] if isinstance(data, dict) else []
|
|
rows = [
|
|
{
|
|
"id": r.get("id"),
|
|
"name": r.get("name"),
|
|
"event": r.get("event"),
|
|
"status": r.get("status"),
|
|
"conclusion": r.get("conclusion"),
|
|
"created_at": r.get("created_at")
|
|
}
|
|
for r in runs
|
|
]
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({"available": True, "rows": rows, "count": len(rows), "repo": repo}))]
|
|
|
|
elif name == "crear_issue":
|
|
title = arguments.get("title")
|
|
body = arguments.get("body", "")
|
|
labels = arguments.get("labels", [])
|
|
|
|
# Get label IDs if names are provided
|
|
label_ids = []
|
|
if labels:
|
|
lbl_response = await client.get(f"/api/v1/repos/{owner}/{repo}/labels")
|
|
if lbl_response.status_code == 200:
|
|
all_labels = lbl_response.json()
|
|
label_ids = [l.get("id") for l in all_labels if l.get("name") in labels]
|
|
|
|
payload = {
|
|
"title": title,
|
|
"body": body,
|
|
"labels": label_ids
|
|
}
|
|
|
|
response = await client.post(f"/api/v1/repos/{owner}/{repo}/issues", json=payload)
|
|
response.raise_for_status()
|
|
issue_data = response.json()
|
|
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({
|
|
"success": True,
|
|
"number": issue_data.get("number"),
|
|
"html_url": issue_data.get("html_url"),
|
|
"title": issue_data.get("title")
|
|
}))]
|
|
|
|
elif name == "comentar_issue":
|
|
issue_number = arguments.get("issue_number")
|
|
body = arguments.get("body")
|
|
|
|
payload = {"body": body}
|
|
response = await client.post(f"/api/v1/repos/{owner}/{repo}/issues/{issue_number}/comments", json=payload)
|
|
response.raise_for_status()
|
|
comment_data = response.json()
|
|
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({
|
|
"success": True,
|
|
"id": comment_data.get("id"),
|
|
"html_url": comment_data.get("html_url")
|
|
}))]
|
|
|
|
elif name == "crear_pull_request":
|
|
title = arguments.get("title")
|
|
body = arguments.get("body", "")
|
|
head = arguments.get("head")
|
|
base = arguments.get("base", "main")
|
|
|
|
payload = {
|
|
"title": title,
|
|
"body": body,
|
|
"head": head,
|
|
"base": base
|
|
}
|
|
|
|
response = await client.post(f"/api/v1/repos/{owner}/{repo}/pulls", json=payload)
|
|
response.raise_for_status()
|
|
pr_data = response.json()
|
|
|
|
import json
|
|
return [TextContent(type="text", text=json.dumps({
|
|
"success": True,
|
|
"number": pr_data.get("number"),
|
|
"html_url": pr_data.get("html_url"),
|
|
"title": pr_data.get("title")
|
|
}))]
|
|
|
|
else:
|
|
raise ValueError(f"Unknown tool: {name}")
|
|
|
|
except httpx.HTTPStatusError as exc:
|
|
logger.error(f"Gitea API error: {exc.response.status_code} - {exc.response.text}")
|
|
raise HTTPException(status_code=exc.response.status_code, detail=f"Gitea API error: {exc.response.text}")
|
|
except Exception as exc:
|
|
logger.error(f"Unexpected error: {str(exc)}")
|
|
raise HTTPException(status_code=500, detail=str(exc))
|
|
|
|
# Initialize SSE Transport
|
|
sse_transport = SseServerTransport("/mcp/messages")
|
|
|
|
@app.get("/mcp/sse")
|
|
async def handle_sse(request: Request):
|
|
"""MCP SSE connection endpoint."""
|
|
async with sse_transport.connect_sse(request.scope, request.receive, request._send) as sse:
|
|
await mcp_server.handle_request_loop(sse.reader, sse.writer)
|
|
|
|
@app.post("/mcp/messages")
|
|
async def handle_messages(request: Request):
|
|
"""MCP messages endpoint for POST requests."""
|
|
await sse_transport.handle_post_message(request.scope, request.receive, request._send)
|
|
|
|
# A2A Agent Card Endpoint
|
|
@app.get("/.well-known/agent.json")
|
|
async def get_agent_card(request: Request):
|
|
"""Serves the Agent Card (A2A manifest) with dynamically resolved URLs."""
|
|
base_url = str(request.base_url).rstrip("/")
|
|
|
|
card = {
|
|
"name": "Gitea Adapter",
|
|
"description": "MCP adapter for Gitea API REST. Wraps Gitea operations as an MCP server.",
|
|
"version": "1.0.0",
|
|
"url": base_url,
|
|
"capabilities": {
|
|
"streaming": True
|
|
},
|
|
"skills": [
|
|
{
|
|
"id": "issues_abiertos",
|
|
"name": "Issues abiertos",
|
|
"description": "Listar issues abiertos en un repositorio de Gitea",
|
|
"tags": ["git", "gitea"]
|
|
},
|
|
{
|
|
"id": "commits_recientes",
|
|
"name": "Commits recientes",
|
|
"description": "Listar commits recientes en una rama",
|
|
"tags": ["git", "gitea"]
|
|
},
|
|
{
|
|
"id": "estado_actions",
|
|
"name": "Estado de Actions",
|
|
"description": "Obtener el estado de las últimas ejecuciones de Gitea Actions",
|
|
"tags": ["git", "gitea", "ci"]
|
|
},
|
|
{
|
|
"id": "crear_issue",
|
|
"name": "Crear issue",
|
|
"description": "Crear un nuevo issue en un repositorio",
|
|
"tags": ["git", "gitea"]
|
|
},
|
|
{
|
|
"id": "comentar_issue",
|
|
"name": "Comentar issue",
|
|
"description": "Agregar un comentario a un issue existente",
|
|
"tags": ["git", "gitea"]
|
|
},
|
|
{
|
|
"id": "crear_pull_request",
|
|
"name": "Crear pull request",
|
|
"description": "Crear un nuevo pull request en un repositorio",
|
|
"tags": ["git", "gitea"]
|
|
}
|
|
],
|
|
"authentication": {
|
|
"schemes": ["bearer"]
|
|
},
|
|
"extensions": {
|
|
"mcp": {
|
|
"endpoint": f"{base_url}/mcp/sse",
|
|
"transport": "streamable-http"
|
|
},
|
|
"capataz": {
|
|
"glyph": "GT",
|
|
"color": "#fc8019"
|
|
}
|
|
}
|
|
}
|
|
return JSONResponse(content=card)
|
|
|
|
# Root status endpoint
|
|
@app.get("/")
|
|
async def root():
|
|
return {
|
|
"status": "online",
|
|
"service": "Gitea MCP Adapter",
|
|
"endpoints": {
|
|
"agent_card": "/.well-known/agent.json",
|
|
"mcp_sse": "/mcp/sse",
|
|
"mcp_messages": "/mcp/messages"
|
|
}
|
|
}
|
|
|
|
if __name__ == "__main__":
|
|
import uvicorn
|
|
port = int(os.getenv("PORT", 8000))
|
|
host = os.getenv("HOST", "0.0.0.0")
|
|
uvicorn.run("main:app", host=host, port=port, reload=True)
|