generated from personal-projects/leo-claude-mktplace
Remove incorrect standalone MCP implementation
This commit removes the incorrectly structured standalone MCP server that was built without understanding the distinction between standalone and HTTP transport modes. These files will be replaced with proper HTTP transport wrapper components. Removed: - src/gitea_mcp/ directory (standalone server implementation) - tests/ directory (tests for standalone implementation) This clears the way for implementing the correct HTTP-wrapped architecture. Closes #9 Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
@@ -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)]
|
||||
Reference in New Issue
Block a user