Acemcp

Instructions

Implementation Reference

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

  Primary handler function for the 'search_context' MCP tool. Validates input arguments, initializes shared IndexManager, calls search_context on it, and formats the result as MCP 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(), defining its metadata, description, and full input schema.
  @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": """Provide a clear natural language description of the code behavior, workflow, or issue you want to locate. You may also add optional keywords to improve semantic matching. Recommended format: Natural language description + optional keywords Examples: “I want to find where the server handles chunk merging in the file upload process. Keywords: upload chunk merge, file service” “Locate where the system refreshes cached data after user permissions are updated. Keywords: permission update, cache refresh” “Find the initialization flow of message queue consumers during startup. Keywords: mq consumer init, subscribe” “Show me how configuration hot-reload is triggered and applied in the code. Keywords: config reload, hot update”""", }, }, "required": ["project_root_path", "query"], }, ), ]
• src/acemcp/server.py:65-82 (registration)

  MCP server dispatcher for tool calls via @app.call_tool(), routing 'search_context' invocations to 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:695-788 (helper)

  Core helper method in IndexManager that performs automatic incremental indexing followed by semantic search via external API, excluding failed blobs, and returns formatted code context snippets.
  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 and exclude failed blobs projects = self._load_projects() all_blob_names = projects.get(normalized_path, []) if not all_blob_names: return f"Error: No blobs found for project {normalized_path} after indexing." # Get failed blob hashes and exclude them from search failed_blob_hashes = self._get_failed_blob_hashes(normalized_path) blob_names = [blob_hash for blob_hash in all_blob_names if blob_hash not in failed_blob_hashes] excluded_count = len(all_blob_names) - len(blob_names) if excluded_count > 0: logger.info(f"Excluded {excluded_count} failed blobs from search (total available: {len(all_blob_names)}, searching: {len(blob_names)})") if not blob_names: return f"Error: No valid blobs available for search in project {normalized_path}. All {len(all_blob_names)} blobs have failed upload." # Step 3: Perform search logger.info(f"Performing search with {len(blob_names)} blobs (excluded {excluded_count} failed 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}"
• src/acemcp/tools/search_context.py:15-39 (helper)

  Shared IndexManager singleton factory used by the tool handler to lazily initialize the index manager with configuration.
  async def _get_index_manager() -> IndexManager: """Create or return the shared IndexManager instance.""" global _index_manager, _index_manager_lock if _index_manager is not None: return _index_manager if _index_manager_lock is None: _index_manager_lock = asyncio.Lock() async with _index_manager_lock: if _index_manager is None: config = get_config() _index_manager = IndexManager( config.index_storage_path, config.base_url, config.token, config.text_extensions, config.batch_size, config.max_lines_per_blob, config.exclude_patterns, ) return _index_manager