feat: MCP write tools -- add_repository, index, directories, delete (OPE-165)

4 new MCP tools so we can manage repos without leaving the conversation:

- add_repository: POST /repos (name, git_url, branch)
- get_repo_directories: GET /repos/{repo_id}/directories
- index_repository: POST /repos/{repo_id}/index (with optional include_paths)
- delete_repository: DELETE /repos/{repo_id}

Also adds api_delete to the HTTP client.
All 45 tests pass, lint clean.

Author: DevanshuNEU

mcp-server/api_client.py

@@ -58,6 +58,14 @@ async def api_post(path: str, json: dict, **kwargs: Any) -> dict:
     return response.json()
 
 
+async def api_delete(path: str, **kwargs: Any) -> dict:
+    """Make a DELETE request to the backend API."""
+    client = await get_client()
+    response = await client.delete(path, **kwargs)
+    response.raise_for_status()
+    return response.json()
+
+
 async def close_client() -> None:
     """Close the persistent client. Call on server shutdown."""
     global _client

mcp-server/handlers.py

@@ -12,7 +12,7 @@
 
 logger = logging.getLogger(__name__)
 
-from api_client import api_get, api_post
+from api_client import api_get, api_post, api_delete
 from formatters import (
     format_codebase_dna,
     format_code_style,
@@ -79,6 +79,90 @@ async def _handle_dna(args: dict[str, Any]) -> str:
     return format_codebase_dna(result)
 
 
+# --- Write tool handlers ---
+
+async def _handle_add_repository(args: dict[str, Any]) -> str:
+    payload = {
+        "name": args["name"],
+        "git_url": args["git_url"],
+        "branch": args.get("branch", "main"),
+    }
+    result = await api_post("/repos", json=payload)
+    repo_id = result.get("id", "unknown")
+    name = result.get("name", args["name"])
+    status = result.get("status", "added")
+    needs_selection = result.get("needs_directory_selection", False)
+    lines = [
+        f"Repository '{name}' added successfully.",
+        f"ID: `{repo_id}`",
+        f"Status: {status}",
+    ]
+    if needs_selection:
+        lines.append(
+            "\nThis repo may benefit from subset indexing. "
+            "Use get_repo_directories to see available directories, "
+            "then index_repository with include_paths."
+        )
+    else:
+        lines.append(
+            f"\nReady to index. Run: index_repository(repo_id='{repo_id}')"
+        )
+    return "\n".join(lines)
+
+
+async def _handle_get_repo_directories(args: dict[str, Any]) -> str:
+    result = await api_get(f"/repos/{args['repo_id']}/directories")
+    dirs = result.get("directories", [])
+    if not dirs:
+        return "No directories found (repo may be flat or not yet cloned)."
+    lines = ["# Repository Directories\n"]
+    for d in dirs:
+        name = d.get("name", d.get("path", "unknown"))
+        count = d.get("file_count", 0)
+        lines.append(f"- **{name}/** -- {count} code files")
+    lines.append(
+        "\nTo index specific directories, use index_repository "
+        "with include_paths=['dir1', 'dir2']."
+    )
+    return "\n".join(lines)
+
+
+async def _handle_index_repository(args: dict[str, Any]) -> str:
+    repo_id = args["repo_id"]
+    include_paths = args.get("include_paths")
+    # Build query params
+    params = {}
+    if include_paths:
+        params["include_paths"] = include_paths
+    result = await api_post(
+        f"/repos/{repo_id}/index",
+        json=params if params else {},
+    )
+    status = result.get("status", "unknown")
+    fn_count = result.get("function_count", result.get("functions_indexed", 0))
+    file_count = result.get("file_count", result.get("files_indexed", 0))
+    lines = [
+        "Indexing complete.",
+        f"Status: {status}",
+        f"Files indexed: {file_count}",
+        f"Functions extracted: {fn_count}",
+    ]
+    if include_paths:
+        lines.append(f"Subset: {', '.join(include_paths)}")
+    lines.append(
+        f"\nYou can now use search_code(repo_id='{repo_id}') "
+        "to search this codebase."
+    )
+    return "\n".join(lines)
+
+
+async def _handle_delete_repository(args: dict[str, Any]) -> str:
+    repo_id = args["repo_id"]
+    result = await api_delete(f"/repos/{repo_id}")
+    msg = result.get("message", "Repository deleted.")
+    return f"{msg}\nRepo ID `{repo_id}` has been removed."
+
+
 # Tool name -> handler mapping
 _HANDLERS: dict[str, Any] = {
     "search_code": _handle_search,
@@ -88,6 +172,10 @@ async def _handle_dna(args: dict[str, Any]) -> str:
     "analyze_impact": _handle_impact,
     "get_repository_insights": _handle_insights,
     "get_codebase_dna": _handle_dna,
+    "add_repository": _handle_add_repository,
+    "get_repo_directories": _handle_get_repo_directories,
+    "index_repository": _handle_index_repository,
+    "delete_repository": _handle_delete_repository,
 }
 
 

mcp-server/tests/test_tools.py

@@ -15,6 +15,10 @@
     "analyze_impact",
     "get_repository_insights",
     "get_codebase_dna",
+    "add_repository",
+    "get_repo_directories",
+    "index_repository",
+    "delete_repository",
 }
 
 

mcp-server/tools.py

@@ -145,4 +145,96 @@ def get_tool_schemas() -> list[types.Tool]:
                 "required": ["repo_id"],
             },
         ),
+        # --- Write tools ---
+        types.Tool(
+            name="add_repository",
+            description=(
+                "Add a new repository for indexing. Clones the repo and analyzes "
+                "its structure. After adding, use get_repo_directories to see "
+                "available directories, then index_repository to start indexing."
+            ),
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "git_url": {
+                        "type": "string",
+                        "description": (
+                            "Git clone URL. "
+                            "Example: https://github.com/owner/repo.git"
+                        ),
+                    },
+                    "name": {
+                        "type": "string",
+                        "description": "Short name for the repository",
+                    },
+                    "branch": {
+                        "type": "string",
+                        "description": "Branch to clone (default: main)",
+                        "default": "main",
+                    },
+                },
+                "required": ["git_url", "name"],
+            },
+        ),
+        types.Tool(
+            name="get_repo_directories",
+            description=(
+                "List top-level directories in a cloned repository with file "
+                "counts. Use this after add_repository to decide which "
+                "directories to index (useful for monorepos)."
+            ),
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "repo_id": {
+                        "type": "string",
+                        "description": "Repository identifier",
+                    }
+                },
+                "required": ["repo_id"],
+            },
+        ),
+        types.Tool(
+            name="index_repository",
+            description=(
+                "Trigger indexing for a repository. Extracts functions, builds "
+                "embeddings, and enables semantic search. For monorepos, pass "
+                "include_paths to index only specific directories."
+            ),
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "repo_id": {
+                        "type": "string",
+                        "description": "Repository identifier",
+                    },
+                    "include_paths": {
+                        "type": "array",
+                        "items": {"type": "string"},
+                        "description": (
+                            "Optional list of directories to index "
+                            "(e.g. ['src', 'lib']). Omit to index everything."
+                        ),
+                    },
+                },
+                "required": ["repo_id"],
+            },
+        ),
+        types.Tool(
+            name="delete_repository",
+            description=(
+                "Delete a repository and all its indexed data. This is "
+                "irreversible -- the repo must be re-added and re-indexed."
+            ),
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "repo_id": {
+                        "type": "string",
+                        "description": "Repository identifier",
+                    }
+                },
+                "required": ["repo_id"],
+            },
+        ),
     ]