diff --git a/src/gitea_mcp/server.py b/src/gitea_mcp/server.py index 2acae7b..63f7755 100644 --- a/src/gitea_mcp/server.py +++ b/src/gitea_mcp/server.py @@ -9,6 +9,14 @@ 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 @@ -35,10 +43,15 @@ async def serve() -> None: """List available MCP tools. Returns: - list: Available tools (placeholder for future implementation). + list: Available tools including issue, label, and milestone operations. """ - # Placeholder tools - will be implemented in issues #3, #4, #5 - return [ + # 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)", @@ -53,32 +66,6 @@ async def serve() -> None: "required": ["org"], }, ), - Tool( - name="create_issue", - description="Create a new issue in a repository (coming soon)", - inputSchema={ - "type": "object", - "properties": { - "owner": { - "type": "string", - "description": "Repository owner", - }, - "repo": { - "type": "string", - "description": "Repository name", - }, - "title": { - "type": "string", - "description": "Issue title", - }, - "body": { - "type": "string", - "description": "Issue body", - }, - }, - "required": ["owner", "repo", "title"], - }, - ), Tool( name="create_pull_request", description="Create a new pull request (coming soon)", @@ -109,7 +96,9 @@ async def serve() -> None: "required": ["owner", "repo", "title", "head", "base"], }, ), - ] + ]) + + return tools @server.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: @@ -122,11 +111,29 @@ async def serve() -> None: Returns: list: Tool response. """ - # Placeholder implementation - actual tools will be implemented in future issues + # 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. Coming soon in issues #3, #4, #5.", + text=f"Tool '{name}' is not yet implemented.", ) ] diff --git a/src/gitea_mcp/tools/__init__.py b/src/gitea_mcp/tools/__init__.py index 6052e3c..f071371 100644 --- a/src/gitea_mcp/tools/__init__.py +++ b/src/gitea_mcp/tools/__init__.py @@ -1 +1,14 @@ """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", +] diff --git a/src/gitea_mcp/tools/issues.py b/src/gitea_mcp/tools/issues.py new file mode 100644 index 0000000..a742109 --- /dev/null +++ b/src/gitea_mcp/tools/issues.py @@ -0,0 +1,392 @@ +"""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)] diff --git a/src/gitea_mcp/tools/labels.py b/src/gitea_mcp/tools/labels.py new file mode 100644 index 0000000..0f8a286 --- /dev/null +++ b/src/gitea_mcp/tools/labels.py @@ -0,0 +1,170 @@ +"""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}", + ) + ] diff --git a/src/gitea_mcp/tools/milestones.py b/src/gitea_mcp/tools/milestones.py new file mode 100644 index 0000000..ab49a72 --- /dev/null +++ b/src/gitea_mcp/tools/milestones.py @@ -0,0 +1,190 @@ +"""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)]