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)