Acemcp

Instructions

Implementation Reference

• src/acemcp/tools/search_context.py:54-86 (handler)

  Primary handler function for executing the 'search_context' tool. Extracts arguments, performs basic validation, initializes shared IndexManager, calls its search_context method, and returns the result as MCP text content.
  async def search_context_tool(arguments: dict[str, Any]) -> dict[str, Any]: """Search for code context based on query. Args: arguments: Tool arguments containing: - project_root_path: Absolute path to the project root directory - query: Search query string Returns: Dictionary containing search results """ try: project_root_path = arguments.get("project_root_path") query = arguments.get("query") if not project_root_path: return {"type": "text", "text": "Error: project_root_path is required"} if not query: return {"type": "text", "text": "Error: query is required"} logger.info(f"Tool invoked: search_context for project {project_root_path} with query: {query}") index_manager = await _get_index_manager() result = await index_manager.search_context(project_root_path, query) return {"type": "text", "text": result} except Exception as e: logger.exception("Error in search_context_tool") return {"type": "text", "text": f"Error: {e!s}"}
• src/acemcp/server.py:24-62 (registration)

  MCP server registration of the 'search_context' tool via @app.list_tools(), including full name, description, and inputSchema defining required parameters project_root_path and query.
  @app.list_tools() async def list_tools() -> list[Tool]: """List available MCP tools. Returns: List of available tools """ return [ Tool( name="search_context", description="Search for relevant code context based on a query within a specific project. " "This tool automatically performs incremental indexing before searching, " "ensuring results are always up-to-date. " "Returns formatted text snippets from the codebase that are semantically related to your query. " "IMPORTANT: Use forward slashes (/) as path separators in project_root_path, even on Windows.", inputSchema={ "type": "object", "properties": { "project_root_path": { "type": "string", "description": "Absolute path to the project root directory. Use forward slashes (/) as separators. Example: C:/Users/username/projects/myproject", }, "query": { "type": "string", "description": "Natural language search query to find relevant code context. This tool performs" " semantic search and returns code snippets that match your query. Examples: " "'logging configuration setup initialization logger' (finds logging setup code), " "'user authentication login' (finds auth-related code), 'database connection pool' " "(finds DB connection code), 'error handling exception' (finds error handling patterns), " "'API endpoint routes' (finds API route definitions). " "The tool returns formatted text snippets with file paths and line numbers showing where the relevant code is located.", }, }, "required": ["project_root_path", "query"], }, ), ]
• src/acemcp/server.py:64-82 (handler)

  MCP server tool dispatcher that checks if name == 'search_context' and invokes the search_context_tool handler.
  @app.call_tool() async def call_tool(name: str, arguments: dict) -> dict: """Handle tool calls. Args: name: Tool name arguments: Tool arguments Returns: Tool execution results """ logger.info(f"Tool called: {name} with arguments: {arguments}") if name == "search_context": return await search_context_tool(arguments) return {"type": "text", "text": f"Unknown tool: {name}"}
• src/acemcp/index/manager.py:488-570 (helper)

  Core helper method implementing the search logic: auto-indexes the project incrementally, loads blob names, sends semantic search request to API (/agents/codebase-retrieval), and formats/retrieves results.
  async def search_context(self, project_root_path: str, query: str) -> str: """Search for code context based on query with automatic incremental indexing. This method automatically performs incremental indexing before searching, ensuring the search is always performed on the latest codebase. Args: project_root_path: Absolute path to the project root directory query: Search query string Returns: Formatted retrieval result """ normalized_path = self._normalize_path(project_root_path) logger.info(f"Searching context in project {normalized_path} with query: {query}") try: # Step 1: Automatically perform incremental indexing logger.info(f"Auto-indexing project {normalized_path} before search...") index_result = await self.index_project(project_root_path) if index_result["status"] == "error": return f"Error: Failed to index project before search. {index_result['message']}" # Log indexing stats if "stats" in index_result: stats = index_result["stats"] logger.info(f"Auto-indexing completed: total={stats['total_blobs']}, existing={stats['existing_blobs']}, new={stats['new_blobs']}") # Step 2: Load indexed blob names projects = self._load_projects() blob_names = projects.get(normalized_path, []) if not blob_names: return f"Error: No blobs found for project {normalized_path} after indexing." # Step 3: Perform search logger.info(f"Performing search with {len(blob_names)} blobs...") payload = { "information_request": query, "blobs": { "checkpoint_id": None, "added_blobs": blob_names, "deleted_blobs": [], }, "dialog": [], "max_output_length": 0, "disable_codebase_retrieval": False, "enable_commit_retrieval": False, } client = self._get_client() async def search_request(): response = await client.post( f"{self.base_url}/agents/codebase-retrieval", headers={"Authorization": f"Bearer {self.token}"}, json=payload, ) response.raise_for_status() return response.json() # Retry up to 3 times with exponential backoff try: result = await self._retry_request(search_request, max_retries=3, retry_delay=2.0) except Exception as e: logger.error(f"Search request failed after retries: {e}") return f"Error: Search request failed after 3 retries. {e!s}" formatted_retrieval = result.get("formatted_retrieval", "") if not formatted_retrieval: logger.warning(f"Search returned empty result for project {normalized_path}") return "No relevant code context found for your query." logger.info(f"Search completed for project {normalized_path}") return formatted_retrieval except Exception as e: logger.exception(f"Failed to search context in project {normalized_path}") return f"Error: {e!s}"