feat: Create issue tool

.devrev/repo.yml

@@ -0,0 +1 @@
+deployable: true

.github/CODEOWNERS

@@ -0,0 +1 @@
+* @kpsunil97

README.md

@@ -8,6 +8,10 @@ A Model Context Protocol server for DevRev. It is used to search and retrieve in
 
 - `search`: Search for information using the DevRev search API with the provided query and namespace.
 - `get_object`: Get all information about a DevRev issue or ticket using its ID.
+- `create_issue`: Creates a DevRev issue with a specified title and part. Parameters:
+  - `title`: The issue title (required)
+  - `part_name`: Name of the part to associate with the issue (required).
+  - The issue will be automatically assigned to the currently authenticated user
 
 ## Configuration
 

src/devrev_mcp/server.py

@@ -14,7 +14,7 @@
 from mcp.server import NotificationOptions, Server
 from pydantic import AnyUrl
 import mcp.server.stdio
-from .utils import make_devrev_request
+from .utils import make_devrev_request, search_part_by_name, get_current_user_id
 
 server = Server("devrev_mcp")
 
@@ -47,6 +47,18 @@ async def handle_list_tools() -> list[types.Tool]:
                 },
                 "required": ["id"],
             },
+        ),
+        types.Tool(
+            name="create_issue",
+            description="Create a DevRev issue with the specified title. The part_name parameter is used to search for and identify the appropriate part ID, and the issue will be automatically assigned to the currently authenticated user.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "title": {"type": "string"},
+                    "part_name": {"type": "string"},
+                },
+                "required": ["title", "part_name"],
+            },
         )
     ]
 
@@ -118,6 +130,65 @@ async def handle_call_tool(
                 text=f"Object information for '{id}':\n{object_info}"
             )
         ]
+    elif name == "create_issue":
+        if not arguments:
+            raise ValueError("Missing arguments")
+
+        title = arguments.get("title")
+        part_name = arguments.get("part_name")
+
+        if not title:
+            raise ValueError("Missing title parameter")
+
+        if not part_name:
+            raise ValueError("Missing part_name parameter")
+
+        # Search for part by name
+        success, part_id, error_message = search_part_by_name(part_name)
+        if not success:
+            return [
+                types.TextContent(
+                    type="text",
+                    text=error_message
+                )
+            ]
+
+        # Get current user ID
+        success, user_id, error_message = get_current_user_id()
+        if not success:
+            return [
+                types.TextContent(
+                    type="text",
+                    text=error_message
+                )
+            ]
+
+        response = make_devrev_request(
+            "works.create",
+            {
+                "type": "issue",
+                "title": title,
+                "applies_to_part": part_id,
+                "owned_by": [user_id]
+            }
+        )
+
+        if response.status_code != 201:
+            error_text = response.text
+            return [
+                types.TextContent(
+                    type="text",
+                    text=f"Create issue failed with status {response.status_code}: {error_text}"
+                )
+            ]
+
+        issue_info = response.json()
+        return [
+            types.TextContent(
+                type="text",
+                text=f"Issue created successfully with ID: {issue_info['display_id']} Info: {issue_info}"
+            )
+        ]
     else:
         raise ValueError(f"Unknown tool: {name}")
 

src/devrev_mcp/utils.py

@@ -7,7 +7,7 @@
 
 import os
 import requests
-from typing import Any, Dict
+from typing import Any, Dict, Tuple, Optional
 
 def make_devrev_request(endpoint: str, payload: Dict[str, Any]) -> requests.Response:
     """
@@ -37,3 +37,56 @@ def make_devrev_request(endpoint: str, payload: Dict[str, Any]) -> requests.Resp
         headers=headers,
         json=payload
     ) 
+
+def search_part_by_name(part_name: str) -> Tuple[bool, Optional[str], Optional[str]]:
+    """
+    Search for a part by name and return its ID if found.
+
+    Args:
+        part_name: The name of the part to search for
+
+    Returns:
+        Tuple containing:
+            - bool: Success status
+            - Optional[str]: Part ID if found, None otherwise
+            - Optional[str]: Error message if there was an error, None otherwise
+    """
+    try:
+        search_response = make_devrev_request(
+            "search.hybrid",
+            {"query": part_name, "namespace": "part"}
+        )
+        if search_response.status_code != 200:
+            return False, None, f"Search for part failed with status {search_response.status_code}: {search_response.text}"
+
+        search_results = search_response.json()
+        if not search_results.get("results") or len(search_results.get("results")) == 0:
+            return False, None, f"No parts found matching '{part_name}'"
+
+        part_id = search_results.get("results")[0].get("part").get("id")
+        return True, part_id, None
+    except Exception as e:
+        return False, None, f"Failed to search for part: {str(e)}"
+
+def get_current_user_id() -> Tuple[bool, Optional[str], Optional[str]]:
+    """
+    Get the ID of the current authenticated user.
+
+    Returns:
+        Tuple containing:
+            - bool: Success status
+            - Optional[str]: User ID if successful, None otherwise
+            - Optional[str]: Error message if there was an error, None otherwise
+    """
+    try:
+        owned_by = make_devrev_request(
+            "dev-users.self",
+            {}
+        )
+        if owned_by.status_code != 200:
+            return False, None, f"Get user failed with status {owned_by.status_code}: {owned_by.text}"
+
+        user_id = owned_by.json().get("dev_user").get("id")
+        return True, user_id, None
+    except Exception as e:
+        return False, None, f"Failed to get current user: {str(e)}"