diff --git a/src/gitea_mcp/server.py b/src/gitea_mcp/server.py index 2acae7b..725fb4e 100644 --- a/src/gitea_mcp/server.py +++ b/src/gitea_mcp/server.py @@ -9,6 +9,7 @@ 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 # Global client instance @@ -35,10 +36,13 @@ async def serve() -> None: """List available MCP tools. Returns: - list: Available tools (placeholder for future implementation). + list: Available tools including issue operations. """ - # Placeholder tools - will be implemented in issues #3, #4, #5 - return [ + # Get issue tools + tools = get_issue_tools() + + # Placeholder for future tools (PR tools, etc.) + tools.extend([ Tool( name="list_repositories", description="List repositories in an organization (coming soon)", @@ -53,32 +57,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 +87,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 +102,17 @@ 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) + + # 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..9fe3724 100644 --- a/src/gitea_mcp/tools/__init__.py +++ b/src/gitea_mcp/tools/__init__.py @@ -1 +1,8 @@ """Gitea MCP tools package.""" + +from .issues import get_issue_tools, handle_issue_tool + +__all__ = [ + "get_issue_tools", + "handle_issue_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)]