diff --git a/main.py b/main.py new file mode 100644 index 0000000..727a7b2 --- /dev/null +++ b/main.py @@ -0,0 +1,419 @@ +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)