init: add main.py implementing MCP and A2A Agent Card
This commit is contained in:
@@ -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)
|
||||
Reference in New Issue
Block a user