MCP Build Environment Service

Instructions

Implementation Reference

• src/server.py:552-627 (handler)

  Primary MCP tool handler for read_file. Extracts repo, path, optional line range and branch from arguments, gets the repository worktree, validates the file path, reads the file content, applies line range if specified, and returns formatted output with line numbers using TextContent.
  async def handle_read_file(self, args: Dict[str, Any]) -> List[TextContent]: """Handle read_file command""" repo = args.get("repo") branch = args.get("branch") file_path = args.get("path") start_line = args.get("start_line") end_line = args.get("end_line") if not file_path: raise ValueError("File path is required") # Get working path with locking lock_key = self._get_lock_key(repo, branch) async with self.worktree_locks[lock_key]: repo_path = await self.get_working_path(repo, branch) # Validate and resolve the file path validated_path = validate_file_path(file_path, repo_path) # Read the file try: with open(validated_path, 'r', encoding='utf-8', errors='replace') as f: if start_line is not None or end_line is not None: # Read specific line range lines = f.readlines() total_lines = len(lines) # Validate line numbers if start_line is not None and start_line < 1: raise ValueError(f"start_line must be >= 1, got {start_line}") if end_line is not None and end_line < 1: raise ValueError(f"end_line must be >= 1, got {end_line}") if start_line is not None and end_line is not None and start_line > end_line: raise ValueError(f"start_line ({start_line}) must be <= end_line ({end_line})") # Default values start_idx = (start_line - 1) if start_line is not None else 0 end_idx = end_line if end_line is not None else total_lines # Clamp to valid range start_idx = max(0, min(start_idx, total_lines)) end_idx = max(0, min(end_idx, total_lines)) # Extract the requested lines selected_lines = lines[start_idx:end_idx] # Format output with line numbers output = f"File: {validated_path}\n" output += f"Lines {start_idx + 1}-{end_idx} of {total_lines}\n" output += "=" * 80 + "\n" for i, line in enumerate(selected_lines, start=start_idx + 1): output += f"{i:6d}: {line.rstrip()}\n" return [TextContent(type="text", text=output)] else: # Read entire file content = f.read() lines_count = content.count('\n') + (1 if content and not content.endswith('\n') else 0) output = f"File: {validated_path}\n" output += f"Total lines: {lines_count}\n" output += "=" * 80 + "\n" # Add line numbers to entire file for i, line in enumerate(content.splitlines(), start=1): output += f"{i:6d}: {line}\n" return [TextContent(type="text", text=output)] except FileNotFoundError: raise FileNotFoundError(f"File not found: {validated_path}") except PermissionError: raise PermissionError(f"Permission denied reading file: {validated_path}") except Exception as e: raise Exception(f"Error reading file {validated_path}: {str(e)}")
• src/server.py:416-447 (schema)

  JSON schema definition for the read_file tool input, returned by list_tools handler. Defines properties repo (required), path (required), start_line/end_line/branch (optional).
  Tool( name="read_file", description="Read the contents of a file in a repository. " "Supports reading specific line ranges for large files. " "If branch is specified, creates/uses a hidden worktree (.repo@branch) for isolation.", inputSchema={ "type": "object", "properties": { "repo": { "type": "string", "description": "Repository name (required)" }, "path": { "type": "string", "description": "Absolute path to the file to read (required)" }, "start_line": { "type": "integer", "description": "Starting line number (1-indexed, optional). If provided, only lines from start_line to end_line will be returned." }, "end_line": { "type": "integer", "description": "Ending line number (1-indexed, optional). If provided, only lines from start_line to end_line will be returned." }, "branch": { "type": "string", "description": "Git branch name (optional). If provided, uses isolated worktree." } }, "required": ["repo", "path"] } )
• src/validators.py:172-242 (helper)

  Helper function validate_file_path used by the read_file handler to ensure the requested file path resolves within the repository root, preventing path traversal attacks, and returns the safe absolute path.
  def validate_file_path(file_path: str, repo_path: Path) -> Path: """ Validate a file path for reading and ensure it's within the repository. This validator accepts both absolute and relative paths but ensures that the final resolved path is within the repository directory. Args: file_path: The file path to validate (can be absolute or relative) repo_path: The repository root path Returns: Path: The validated absolute path to the file Raises: ValueError: If path contains dangerous patterns or escapes the repository """ if not file_path: raise ValueError("File path cannot be empty") # Check for dangerous command injection patterns # We're less strict than validate_path since we're only reading files dangerous_for_file_read = [ r";", # Command chaining r"\|", # Pipes r"&", # Background/chaining r"`", # Command substitution r"\$\(", # Command substitution ] for pattern in dangerous_for_file_read: if re.search(pattern, file_path): raise ValueError(f"File path contains dangerous pattern: {file_path}") # Convert to Path object path_obj = Path(file_path) # Resolve the path to absolute if path_obj.is_absolute(): # Absolute path - resolve it resolved_path = path_obj.resolve() else: # Relative path - resolve relative to repo resolved_path = (repo_path / path_obj).resolve() # Ensure the resolved path is within the repository repo_path_resolved = repo_path.resolve() try: # This will raise ValueError if resolved_path is not relative to repo_path resolved_path.relative_to(repo_path_resolved) except ValueError: raise ValueError( f"Access denied: Path '{file_path}' resolves to '{resolved_path}' " f"which is outside repository '{repo_path_resolved}'" ) # Additional check: ensure no parent directory traversal in original path # This catches things like "foo/../../etc/passwd" even if they resolve safely if ".." in path_obj.parts: # But we need to verify it doesn't escape try: if path_obj.is_absolute(): check_path = path_obj.resolve() else: check_path = (repo_path / path_obj).resolve() check_path.relative_to(repo_path_resolved) except ValueError: raise ValueError(f"Path traversal not allowed: {file_path}") return resolved_path
• src/server.py:165-169 (registration)

  MCP server decorator registration for list_tools handler, which returns the list of tools including read_file with its schema.
  @self.server.list_tools() async def handle_list_tools() -> List[Tool]: """List available MCP tools""" return await self.get_tools_list()
• src/server.py:170-174 (registration)

  MCP server decorator registration for call_tool handler, which dispatches to execute_tool based on name, including read_file.
  @self.server.call_tool() async def handle_call_tool(name: str, arguments: Any) -> List[TextContent]: """Handle tool execution""" return await self.execute_tool(name, arguments)