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

mcp-server/api_client.py

@@ -58,6 +58,16 @@ 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()
+    if response.status_code == 204 or not response.content:
+        return {}
+    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,99 @@ 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("repo_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")
+
+    if include_paths is not None and len(include_paths) == 0:
+        return "Error: include_paths cannot be empty. Omit it to index the full repo, or provide directory names."
+
+    if include_paths:
+        # Async endpoint supports include_paths for monorepo subset indexing
+        result = await api_post(
+            f"/repos/{repo_id}/index/async",
+            json={"include_paths": include_paths},
+        )
+        status = result.get("status", "accepted")
+        return (
+            f"Async indexing started for subset: {', '.join(include_paths)}\n"
+            f"Status: {status}\n"
+            f"Repo ID: `{repo_id}`\n"
+            "\nIndexing runs in the background. Use list_repositories "
+            "to check when status changes to 'indexed'."
+        )
+
+    # Sync endpoint for full-repo indexing
+    result = await api_post(f"/repos/{repo_id}/index", json={})
+    status = result.get("status", "unknown")
+    fn_count = result.get("functions", 0)
+    lines = [
+        "Indexing complete.",
+        f"Status: {status}",
+        f"Functions extracted: {fn_count}",
+    ]
+    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 +181,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",
 }
 
 
@@ -53,6 +57,7 @@ def test_repo_tools_require_repo_id(self):
         repo_tools = [
             "get_dependency_graph", "analyze_code_style",
             "analyze_impact", "get_repository_insights", "get_codebase_dna",
+            "get_repo_directories", "index_repository", "delete_repository",
         ]
         for name in repo_tools:
             required = schemas[name].inputSchema.get("required", [])

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"],
+            },
+        ),
     ]